close()

Properly close database connections and free resources across all supported database backends.

Overview

The close() method properly closes database connections and releases associated resources. This is essential for preventing connection leaks, especially in long-running applications or when working with connection-limited databases.

Signature

close() -> None

📝 Important Notes

  • • Calling close() makes the Akron instance unusable for further database operations
  • • Always close connections when done, especially in production applications
  • • Use context managers (with statements) for automatic connection management
  • • Close operations are idempotent - calling close() multiple times is safe

Parameters

The close() method takes no parameters.

Returns

Type: None

The method does not return a value. It performs cleanup operations and closes the database connection silently.

Examples

Basic Connection Management

Manual Connection Closing
1from akron import Akron
2
3# Create database connection
4db = Akron("sqlite:///example.db")
5
6# Perform database operations
7db.create_table("users", {
8 "id": "int",
9 "username": "str",
10 "email": "str"
11})
12
13user_id = db.insert("users", {"username": "alice", "email": "alice@example.com"})
14print(f"Created user with ID: {user_id}")
15
16# Query the data
17users = db.find("users")
18print(f"Found {len(users)} user(s)")
19
20# Always close the connection when done
21db.close()
22print("Database connection closed")
Expected Output
Created user with ID: 1
Found 1 user(s)
Database connection closed

Context Manager (Recommended)

Use the with statement for automatic connection management:

Automatic Connection Management
1from akron import Akron
2
3# Context manager automatically handles opening and closing
4with Akron("sqlite:///example.db") as db:
5 # Create and populate table
6 db.create_table("products", {
7 "id": "int",
8 "name": "str",
9 "price": "float",
10 "category": "str"
11 })
12
13 # Insert some products
14 products = [
15 {"name": "Laptop", "price": 999.99, "category": "Electronics"},
16 {"name": "Book", "price": 19.99, "category": "Books"},
17 {"name": "Coffee Mug", "price": 12.99, "category": "Kitchen"}
18 ]
19
20 for product in products:
21 product_id = db.insert("products", product)
22 print(f"Added {product['name']} with ID: {product_id}")
23
24 # Query results
25 all_products = db.find("products")
26 print(f"\nTotal products in database: {len(all_products)}")
27
28 # Connection automatically closed when exiting the with block
29
30print("\nConnection closed automatically via context manager")
Expected Output
Added Laptop with ID: 1
Added Book with ID: 2
Added Coffee Mug with ID: 3

Total products in database: 3

Connection closed automatically via context manager

Multiple Database Connections

Manage multiple database connections properly:

Managing Multiple Connections
1from akron import Akron
2
3def sync_data_between_databases():
4 # Open connections to both databases
5 source_db = Akron("sqlite:///source.db")
6 target_db = Akron("sqlite:///target.db")
7
8 try:
9 # Setup source database
10 source_db.create_table("users", {
11 "id": "int",
12 "username": "str",
13 "email": "str",
14 "created_date": "str"
15 })
16
17 # Add sample data
18 source_db.insert("users", {"username": "john", "email": "john@example.com", "created_date": "2024-01-01"})
19 source_db.insert("users", {"username": "jane", "email": "jane@example.com", "created_date": "2024-01-02"})
20
21 # Setup target database with same structure
22 target_db.create_table("users", {
23 "id": "int",
24 "username": "str",
25 "email": "str",
26 "created_date": "str",
27 "synced_date": "str"
28 })
29
30 # Sync data from source to target
31 users = source_db.find("users")
32 synced_count = 0
33
34 for user in users:
35 # Add sync timestamp
36 user["synced_date"] = "2024-01-15"
37 target_db.insert("users", user)
38 synced_count += 1
39
40 print(f"Synced {synced_count} users from source to target database")
41
42 # Verify sync
43 target_users = target_db.find("users")
44 print(f"Target database now has {len(target_users)} users")
45
46 finally:
47 # Always close both connections
48 source_db.close()
49 target_db.close()
50 print("\nBoth database connections closed")
51
52# Run the sync operation
53sync_data_between_databases()
Expected Output
Synced 2 users from source to target database
Target database now has 2 users

Both database connections closed

Error Handling with Proper Cleanup

Ensure connections are closed even when errors occur:

Error-Safe Connection Management
1from akron import Akron
2from akron.exceptions import AkronError
3
4def safe_database_operation():
5 db = None
6 try:
7 db = Akron("sqlite:///example.db")
8
9 # Potentially risky operations
10 db.create_table("orders", {
11 "id": "int",
12 "customer_id": "int",
13 "total": "float",
14 "status": "str"
15 })
16
17 # This might fail due to constraints or validation
18 db.insert("orders", {"customer_id": 1, "total": 150.00, "status": "pending"})
19
20 # Simulate an error condition
21 result = db.find("orders", {"customer_id": 1})
22 if len(result) > 0:
23 print(f"Order created successfully: ID {result[0]['id']}")
24
25 return True
26
27 except AkronError as e:
28 print(f"Database operation failed: {e}")
29 return False
30
31 except Exception as e:
32 print(f"Unexpected error: {e}")
33 return False
34
35 finally:
36 # Always close the connection, even if an error occurred
37 if db is not None:
38 db.close()
39 print("Database connection closed in finally block")
40
41# Alternative: Using context manager for automatic cleanup
42def safe_database_operation_with_context():
43 try:
44 with Akron("sqlite:///example.db") as db:
45 db.create_table("customers", {
46 "id": "int",
47 "name": "str",
48 "email": "str"
49 })
50
51 customer_id = db.insert("customers", {"name": "Alice Smith", "email": "alice@example.com"})
52 print(f"Customer created with ID: {customer_id}")
53
54 # Even if an error occurs here, connection will be closed
55 customers = db.find("customers")
56 print(f"Found {len(customers)} customer(s)")
57
58 except AkronError as e:
59 print(f"Database error: {e}")
60 except Exception as e:
61 print(f"Unexpected error: {e}")
62
63 print("Connection automatically closed by context manager")
64
65# Run both examples
66print("=== Manual cleanup example ===")
67safe_database_operation()
68
69print("\n=== Context manager example ===")
70safe_database_operation_with_context()
Expected Output
=== Manual cleanup example ===
Order created successfully: ID 1
Database connection closed in finally block

=== Context manager example ===
Customer created with ID: 1
Found 1 customer(s)
Connection automatically closed by context manager

Database-Specific Behavior

🗄️ SQLite

Closes the SQLite connection and commits any pending transactions:

1# SQLite close operation
2try:
3 self.connection.close()
4except sqlite3.Error:
5 pass # Ignore errors during close

🐬 MySQL

Closes MySQL connection and releases server resources:

1# MySQL close operation
2try:
3 if self.connection.is_connected():
4 self.connection.close()
5except mysql.connector.Error:
6 pass # Ignore errors during close

🐘 PostgreSQL

Closes PostgreSQL connection and returns it to the connection pool:

1# PostgreSQL close operation
2try:
3 self.connection.close()
4except psycopg2.Error:
5 pass # Ignore errors during close

🍃 MongoDB

Closes MongoDB client connection:

1# MongoDB close operation
2try:
3 self.client.close()
4except pymongo.errors.PyMongoError:
5 pass # Ignore errors during close

Best Practices

🔄 Automatic Management

  • Prefer context managers (with statements) for automatic connection cleanup
  • Use try/finally blocks when manual management is necessary
  • Close connections as soon as you're done with database operations
  • Never leave connections open indefinitely in long-running applications

🛡️ Error Handling

  • Always close connections in finally blocks or exception handlers
  • Don't let exceptions prevent proper connection cleanup
  • Handle close() errors gracefully (connection might already be closed)
  • Log connection management events for debugging

📊 Resource Management

  • Monitor connection counts in production applications
  • Implement connection pooling for high-traffic applications
  • Set appropriate connection timeouts
  • Consider using connection pools for better resource utilization

🧪 Testing

  • Test connection cleanup in unit tests
  • Verify that connections are properly closed after operations
  • Test error scenarios to ensure cleanup still occurs
  • Monitor for connection leaks in integration tests

Common Patterns

Connection Management Patterns
1# Pattern 1: Simple operation with context manager (recommended)
2with Akron("sqlite:///app.db") as db:
3 result = db.find("users", {"active": True})
4 return result
5# Connection automatically closed
6
7# Pattern 2: Manual management with proper cleanup
8def process_data(db_config):
9 db = None
10 try:
11 db = Akron(db_config)
12 # ... database operations ...
13 return results
14 finally:
15 if db:
16 db.close()
17
18# Pattern 3: Reusable database service class
19class DatabaseService:
20 def __init__(self, db_config):
21 self.db_config = db_config
22 self.db = None
23
24 def connect(self):
25 if self.db is None:
26 self.db = Akron(self.db_config)
27
28 def disconnect(self):
29 if self.db:
30 self.db.close()
31 self.db = None
32
33 def __enter__(self):
34 self.connect()
35 return self.db
36
37 def __exit__(self, exc_type, exc_val, exc_tb):
38 self.disconnect()
39
40# Usage:
41with DatabaseService("sqlite:///app.db") as db:
42 users = db.find("users")
43
44# Pattern 4: Multiple operations with single connection
45def batch_operations():
46 with Akron("sqlite:///batch.db") as db:
47 # Multiple operations using same connection
48 db.create_table("logs", {"id": "int", "message": "str", "timestamp": "str"})
49
50 for i in range(100):
51 db.insert("logs", {
52 "message": f"Batch operation {i}",
53 "timestamp": "2024-01-15"
54 })
55
56 # All operations complete before connection closes
57 return db.find("logs")
58
59# Pattern 5: Connection state checking
60def safe_operation(db):
61 try:
62 # Check if connection is still valid
63 db.find("users", limit=1) # Quick test query
64
65 # Proceed with actual operation
66 return db.find("users", {"active": True})
67
68 except AkronError:
69 # Connection might be closed or invalid
70 print("Database connection is not available")
71 return []

Integration Examples

Web Application Pattern
1# Flask application with proper connection management
2from flask import Flask, request, jsonify
3from akron import Akron
4
5app = Flask(__name__)
6DATABASE_URL = "sqlite:///app.db"
7
8@app.route("/users", methods=["GET"])
9def get_users():
10 with Akron(DATABASE_URL) as db:
11 users = db.find("users", {"active": True})
12 return jsonify(users)
13
14@app.route("/users", methods=["POST"])
15def create_user():
16 user_data = request.json
17
18 with Akron(DATABASE_URL) as db:
19 user_id = db.insert("users", user_data)
20 created_user = db.find("users", {"id": user_id})[0]
21 return jsonify(created_user), 201
22
23# Background task with connection management
24import threading
25
26def background_cleanup():
27 with Akron(DATABASE_URL) as db:
28 # Clean up old data
29 deleted = db.delete("logs", {"status": "processed"})
30 print(f"Cleaned up {deleted} processed log entries")
31
32# Schedule cleanup to run periodically
33cleanup_thread = threading.Thread(target=background_cleanup)
34cleanup_thread.daemon = True
35cleanup_thread.start()

Next Steps

Now that you understand connection management, explore:

Getting Started Guide

Learn the basics of setting up and using Akron ORM

→ view getting started guide

Database Support

Learn about database-specific features and configurations

→ view database support