Use before implementing pgdbm patterns to avoid common mistakes - provides rationalization table and red flags that prevent pool multiplication, schema errors, and template syntax violations
Core Principle: Most pgdbm mistakes come from fighting the library's design instead of using it.
This skill provides explicit counters for common rationalizations that lead to bugs.
Violating these = your code is wrong:
drop_test_database() in a finally block| Excuse | Reality | Fix |
|---|---|---|
| "Each service needs different pool sizes" | Shared pool allocates dynamically. Pre-sizing is guessing. | ONE pool with total max |
| "Separate pools give better isolation" | Schema isolation is enough. Separate pools waste connections. | Schema-isolated managers |
| "It's simpler to just write the schema name" | Breaks portability. Code only works in one deployment mode. | Use {{tables.}} always |
| "I'll use search_path instead of templates" | Doesn't work with shared pools. Leads to race conditions. | Use {{tables.}} syntax |
| "module_name seems optional" | Causes migration conflicts when multiple modules share DB. | Always specify unique name |
| "I can switch schema at runtime for tenants" | Race conditions. Same manager used by concurrent requests. | Create manager per schema |
| "I'll close the db_manager in my library" | Closes parent app's pool. Crashes everything. | Check _external_db flag |
| "{{tables.}} is too verbose, I'll skip it" | Works until you use shared pools or change schemas. Then breaks. | Use always, no exceptions |
| "Cleanup doesn't need try/finally" | If test fails, cleanup never runs. Databases leak forever. | ALWAYS use try/finally |
| "I'll silence cleanup errors with except pass" | Hides failures. Databases accumulate silently for months. | Let cleanup errors propagate |
If you're about to do ANY of these, you're making a mistake:
# WRONG
service1_db = AsyncDatabaseManager(DatabaseConfig(connection_string="postgresql://localhost/app"))
service2_db = AsyncDatabaseManager(DatabaseConfig(connection_string="postgresql://localhost/app"))
What happens:
"⚠️ Creating another connection pool to..."Fix:
# CORRECT
pool = await AsyncDatabaseManager.create_shared_pool(config)
service1_db = AsyncDatabaseManager(pool=pool, schema="service1")
service2_db = AsyncDatabaseManager(pool=pool, schema="service2")
# WRONG
await db.execute('INSERT INTO "myschema".users (email) VALUES ($1)', email)
await db.execute('INSERT INTO users (email) VALUES ($1)', email)
What happens:
Fix:
# CORRECT
await db.execute('INSERT INTO {{tables.users}} (email) VALUES ($1)', email)
# WRONG
migrations = AsyncMigrationManager(db, "migrations")
# Uses "default" module name - conflicts with other modules!
What happens:
Fix:
# CORRECT
migrations = AsyncMigrationManager(db, "migrations", module_name="myservice")
# WRONG
migrations = AsyncMigrationManager(
db,
"migrations",
schema="myschema" # This parameter doesn't exist!
)
What happens:
Fix:
# CORRECT - schema comes from db
db = AsyncDatabaseManager(pool=pool, schema="myschema")
migrations = AsyncMigrationManager(db, "migrations", module_name="myservice")
# WRONG
db = AsyncDatabaseManager(pool=pool, schema="tenant1")
# Later...
db.schema = "tenant2" # Don't do this!
await db.execute("INSERT INTO {{tables.data}} ...")
What happens:
Fix:
# CORRECT - create manager per schema
tenant1_db = AsyncDatabaseManager(pool=pool, schema="tenant1")
tenant2_db = AsyncDatabaseManager(pool=pool, schema="tenant2")
# WRONG
db = AsyncDatabaseManager(pool=shared_pool, schema="myservice")
await db.connect() # ERROR!
What happens:
Fix:
# CORRECT - don't call connect() when using external pool
db = AsyncDatabaseManager(pool=shared_pool, schema="myservice")
# Just use it - no connect() needed
# WRONG in library
class MyLibrary:
async def close(self):
# Always disconnects, even if didn't create connection
await self.db.disconnect()
What happens:
Fix:
# CORRECT - conditional cleanup
class MyLibrary:
async def close(self):
if self.db and not self._external_db:
await self.db.disconnect()
# WRONG - inconsistent
await db.execute('CREATE TABLE {{tables.users}} (...)')
await db.execute('INSERT INTO users (email) VALUES ($1)', email)
What happens:
Fix:
# CORRECT - use templates everywhere
await db.execute('CREATE TABLE {{tables.users}} (...)')
await db.execute('INSERT INTO {{tables.users}} (email) VALUES ($1)', email)
# WRONG - cleanup never runs if test fails
@pytest_asyncio.fixture
async def test_db():
test_database = AsyncTestDatabase(TEST_CONFIG)
await test_database.create_test_database()
async with test_database.get_test_db_manager(schema="myapp") as db:
yield db
await test_database.drop_test_database() # ← NEVER RUNS IF TEST FAILS
What happens:
test_* databases accumulate (thousands over time)Fix:
# CORRECT - cleanup in finally block
@pytest_asyncio.fixture
async def test_db():
test_database = AsyncTestDatabase(TEST_CONFIG)
await test_database.create_test_database()
try:
async with test_database.get_test_db_manager(schema="myapp") as db:
yield db
finally:
await test_database.drop_test_database() # ← ALWAYS RUNS
Even better - use provided fixtures:
# BEST - just import and use pgdbm fixtures
# tests/conftest.py
from pgdbm.fixtures.conftest import *
# No manual cleanup needed - fixtures handle it
# WRONG - silently ignores cleanup failure