The prompt #
What it gave me #
import asyncpg
from contextlib import asynccontextmanager
class DatabasePool:
def __init__(self, dsn: str, min_size: int = 10, max_size: int = 50):
self.dsn = dsn
self.min_size = min_size
self.max_size = max_size
self._pool: asyncpg.Pool | None = None
self._health_check_task: asyncio.Task | None = None
async def initialize(self) -> None:
self._pool = await asyncpg.create_pool(
self.dsn,
min_size=self.min_size,
max_size=self.max_size,
command_timeout=60,
health_check_interval=30,
health_check_query="SELECT 1",
)
self._health_check_task = asyncio.create_task(self._health_monitor())
@asynccontextmanager
async def acquire(self) -> AsyncGenerator[asyncpg.Connection, None]:
if self._pool is None:
raise RuntimeError("Pool not initialized")
async with self._pool.acquire() as conn:
yield conn
async def _health_monitor(self) -> None:
while True:
await asyncio.sleep(30)
await self._pool.health_check()
async def close(self) -> None:
if self._health_check_task:
self._health_check_task.cancel()
if self._pool:
await self._pool.close()
The error that sent me down the rabbit hole #
TypeError: create_pool() got unexpected keyword arguments:
'health_check_interval', 'health_check_query'
AttributeError: 'Pool' object has no attribute 'health_check'
What I tried before realizing it was hallucinated #
- Checked asyncpg docs — no
health_check_interval
, no health_check_query
, no Pool.health_check()
-
Searched GitHub issues — zero results for these parameters
-
Dug through source —
create_pool
signature only accepts: dsn
, min_size
, max_size
, command_timeout
, timeout
, loop
, connection_class
, init
, setup
, server_settings
, statement_cache_size
, max_queries
, max_inactive_connection_lifetime
- Tested on 0.28.0, 0.29.0, 0.30.0 — same error every version
Where the hallucination likely came from #
The model conflated three real things:
| Real thing | What it does |
|------------|--------------|
| max_inactive_connection_lifetime
(asyncpg 0.29+) | Closes idle connections after N seconds |
| asyncpg.Pool.acquire()
timeout parameter | Fails fast if no connection available |
| asyncpg.Connection.execute("SELECT 1")
| Manual health check pattern people actually use |
It then invented a cohesive API around these concepts that looks like it should exist — consistent naming, sensible defaults, proper async context manager usage. That's the dangerous part: it's not random garbage, it's plausible garbage.
The actual working pattern #
import asyncpg
import asyncio
from contextlib import asynccontextmanager
class DatabasePool:
def __init__(self, dsn: str, min_size: int = 10, max_size: int = 50):
self.dsn = dsn
self.min_size = min_size
self.max_size = max_size
self._pool: asyncpg.Pool | None = None
async def initialize(self) -> None:
self._pool = await asyncpg.create_pool(
self.dsn,
min_size=self.min_size,
max_size=self.max_size,
command_timeout=60,
max_inactive_connection_lifetime=300, # Real parameter
)
@asynccontextmanager
async def acquire(self) -> AsyncGenerator[asyncpg.Connection, None]:
if self._pool is None:
raise RuntimeError("Pool not initialized")
async with self._pool.acquire(timeout=10) as conn: # Real timeout
yield conn
async def health_check(self) -> bool:
"""Manual health check — call this from your /health endpoint"""
if self._pool is None:
return False
try:
async with self._pool.acquire(timeout=2) as conn:
await conn.execute("SELECT 1")
return True
except Exception:
return False
async def close(self) -> None:
if self._pool:
await self._pool.close()
The pattern I'm seeing #
Hallucinations cluster around plausible API extensions — methods/parameters that should exist based on naming conventions and common patterns. The model isn't retrieving; it's synthesizing a consistent interface from partial knowledge.
Anyone else hit this with asyncpg or other libraries? Wondering if there's a systematic way to catch these before they waste hours.
Next Greg Brockman's tightening grip on OpenAI has me nervous about →
a library of Claude prompt techniques, with plenty of directly applicable cases.