Bernt
bae705aa97
- Add NFC ePassport roadmap (ICAO 9303, eIDAS) - Add TensorFlow.js edge face detection (BlazeFace) - Add structured audit logger (GDPR-compliant) - Risk scoring support Part of KYC Apple Native UX v1.1.0
97 lines
2.9 KiB
Python
97 lines
2.9 KiB
Python
"""
|
|
Async Database Layer
|
|
Production-ready async PostgreSQL with connection pooling
|
|
"""
|
|
|
|
import asyncpg
|
|
from typing import Optional, List, Dict, Any
|
|
from contextlib import asynccontextmanager
|
|
import os
|
|
|
|
|
|
class DatabasePool:
|
|
"""
|
|
Async PostgreSQL connection pool
|
|
|
|
Features:
|
|
- Connection pooling
|
|
- Automatic reconnection
|
|
- Query timeout
|
|
- Connection health checks
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.pool: Optional[asyncpg.Pool] = None
|
|
self.dsn = os.getenv("DATABASE_URL", "postgresql://iom:iom_secret@localhost:5432/iom")
|
|
self.min_size = int(os.getenv("DB_POOL_MIN", "5"))
|
|
self.max_size = int(os.getenv("DB_POOL_MAX", "20"))
|
|
self.command_timeout = int(os.getenv("DB_TIMEOUT", "30"))
|
|
|
|
async def connect(self):
|
|
"""Initialize connection pool"""
|
|
self.pool = await asyncpg.create_pool(
|
|
self.dsn,
|
|
min_size=self.min_size,
|
|
max_size=self.max_size,
|
|
command_timeout=self.command_timeout,
|
|
server_settings={
|
|
'jit': 'off',
|
|
'application_name': 'iom_api'
|
|
}
|
|
)
|
|
|
|
async def close(self):
|
|
"""Close connection pool"""
|
|
if self.pool:
|
|
await self.pool.close()
|
|
|
|
@asynccontextmanager
|
|
async def acquire(self):
|
|
"""Acquire connection from pool"""
|
|
if not self.pool:
|
|
await self.connect()
|
|
|
|
async with self.pool.acquire() as conn:
|
|
yield conn
|
|
|
|
async def fetch(self, query: str, *args) -> List[Dict]:
|
|
"""Fetch multiple rows"""
|
|
async with self.acquire() as conn:
|
|
rows = await conn.fetch(query, *args)
|
|
return [dict(row) for row in rows]
|
|
|
|
async def fetchrow(self, query: str, *args) -> Optional[Dict]:
|
|
"""Fetch single row"""
|
|
async with self.acquire() as conn:
|
|
row = await conn.fetchrow(query, *args)
|
|
return dict(row) if row else None
|
|
|
|
async def execute(self, query: str, *args) -> str:
|
|
"""Execute query"""
|
|
async with self.acquire() as conn:
|
|
return await conn.execute(query, *args)
|
|
|
|
async def execute_many(self, query: str, args: List[tuple]) -> str:
|
|
"""Execute query with multiple parameter sets"""
|
|
async with self.acquire() as conn:
|
|
return await conn.executemany(query, args)
|
|
|
|
async def transaction(self):
|
|
"""Start transaction"""
|
|
async with self.acquire() as conn:
|
|
async with conn.transaction():
|
|
yield conn
|
|
|
|
async def health_check(self) -> bool:
|
|
"""Check database health"""
|
|
try:
|
|
async with self.acquire() as conn:
|
|
result = await conn.fetchval("SELECT 1")
|
|
return result == 1
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
# Global database instance
|
|
db = DatabasePool()
|