ConnectionManager
mydborm.db.ConnectionManager
Central connection manager.
Usage — direct config: db.configure(dialect="mysql", host="localhost", user="root", password="root", database="testdb")
Usage — from environment variable: os.environ["DATABASE_URL"] = "mysql://root:root@localhost:3306/testdb" db.from_env()
Usage — as context manager: with db.connect() as conn: cur = conn.cursor() cur.execute("SELECT 1")
encoding
property
Python encoding for text handling (default utf-8).
bulk_transaction()
Atomic transaction across multiple model operations. ALL operations commit together or ALL roll back together.
Usage
with db.bulk_transaction() as tx: tx.execute("INSERT INTO users ...") tx.execute("INSERT INTO profiles ...") tx.execute("INSERT INTO orders ...")
all committed atomically
The tx object is the connection — use db.execute() inside.
close()
Close the current thread's connection.
configure(**kwargs)
Set connection config directly as keyword arguments.
Other Parameters:
| Name | Type | Description |
|---|---|---|
dialect |
str
|
"mysql" or "yugabyte" |
host |
str
|
database host |
port |
int
|
database port |
user |
str
|
database user |
password |
str
|
database password |
database |
str
|
database name |
charset |
str
|
character set (default utf8mb4 for MySQL) |
encoding |
str
|
python encoding (default utf-8) |
configure_pool(pool_size=5, max_overflow=10, pool_timeout=30, pool_recycle=3600)
Configure connection pool settings.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
pool_size
|
int
|
number of persistent connections (default 5) |
5
|
max_overflow
|
int
|
extra connections allowed above pool_size |
10
|
pool_timeout
|
int
|
seconds to wait for a connection (default 30) |
30
|
pool_recycle
|
int
|
seconds before recycling a connection (default 3600) |
3600
|
Usage
db.configure(dialect="mysql", ...) db.configure_pool(pool_size=10, max_overflow=20)
connect()
Thread-safe connection with automatic commit / rollback.
with db.connect() as conn: cur = conn.cursor() cur.execute("INSERT INTO ...")
auto-committed here
execute(sql, params=None)
Execute a raw SQL statement (INSERT, UPDATE, DELETE, DDL). Returns number of affected rows.
Usage
db.execute("UPDATE users SET active = %s WHERE id = %s", [False, 1])
fetchall(sql, params=None)
Execute a raw SELECT and return list of dicts.
Usage
rows = db.fetchall( "SELECT * FROM users WHERE active = %s", [True] )
fetchone(sql, params=None)
Execute a raw SELECT and return a single row dict or None.
Usage
row = db.fetchone( "SELECT * FROM users WHERE email = %s", ["alice@example.com"] )
from_env(var='DATABASE_URL')
Load config from an environment variable. Works on Windows, Linux, and macOS.
list_tables()
Return list of all table names in the current database.
Usage
tables = db.list_tables() print(tables) # ['users', 'products', ...]
nested_transaction()
Create a nested transaction using savepoints. If already inside a transaction, uses a savepoint. If not, starts a new transaction.
Usage
with db.transaction(): User.create(username="alice") with db.nested_transaction(): User.create(username="bob") # if this fails, only bob rolls back
ping()
Ping the database to check connectivity. Returns True if connected, False otherwise.
Usage
if db.ping(): print("Database is reachable")
pool_status()
Return current pool configuration and connection status.
Usage
status = db.pool_status() print(status)
reconnect()
Force close and reopen the connection. Useful after database restarts or stale connections.
Usage
db.reconnect()
savepoint(name=None)
Create a savepoint within an active transaction. Allows partial rollback without rolling back the entire transaction.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
savepoint name (auto-generated if not provided) |
None
|
Usage
with db.transaction(): User.create(username="alice") with db.savepoint("after_alice"): User.create(username="bob") raise Exception("bob failed") # only bob is rolled back, alice is kept
table_exists(table)
Check if a table exists in the current database.
Usage
if db.table_exists("users"): print("Table exists")
transaction()
Explicit transaction context manager. All statements inside the block are committed together. Any exception triggers a full rollback.
Usage
with db.transaction() as conn: db.execute("INSERT INTO users ...") db.execute("INSERT INTO profiles ...")
both committed or both rolled back
transaction_with_retry(retries=3, retry_delay=0.5)
Transaction that retries on deadlock with exponential backoff. Non-deadlock exceptions are raised immediately without retry.