rdmsandClaude Opus 5.5
979e1c0edc
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
32 lines
848 B
Python
32 lines
848 B
Python
from sqlalchemy import create_engine, text
|
|
from sqlalchemy.orm import DeclarativeBase, Session
|
|
|
|
from app.config import get_settings
|
|
|
|
|
|
class Base(DeclarativeBase):
|
|
pass
|
|
|
|
|
|
_engine = None
|
|
|
|
|
|
def get_engine():
|
|
global _engine
|
|
if _engine is None:
|
|
_engine = create_engine(get_settings().database_url)
|
|
return _engine
|
|
|
|
|
|
def get_session():
|
|
with Session(get_engine()) as session:
|
|
yield session
|
|
|
|
|
|
def advisory_xact_lock(session: Session, key: int) -> None:
|
|
"""Serialisiert konkurrierende Schreiber abgeleiteter Daten (Projektionen,
|
|
Planungs-Diagramm) über einen Postgres-Transaktions-Advisory-Lock. Der
|
|
Lock endet mit dem nächsten commit/rollback. Auf SQLite (Tests) No-Op."""
|
|
if session.get_bind().dialect.name == "postgresql":
|
|
session.execute(text("SELECT pg_advisory_xact_lock(:k)"), {"k": key})
|