snippet.py

# db.py

engine = create_engine(
  db_config.url,
  echo=bool(os.getenv("DEBUG", False)),
  hide_parameters=not is_pytest(),
  pool_size=db_config.pool_size,
  pool_pre_ping=True,
  pool_timeout=db_config.pool_checkout_timeout_seconds,
  max_overflow=math.ceil(db_config.pool_size * db_config.pool_overflow_fraction),
  future=True,
  connect_args=get_connection_options(db_config),
)

SessionLocal = sessionmaker(
    expire_on_commit=False,
    autocommit=False,
    autoflush=False,
    bind=engine,
    future=True,
)

def get_raw_readonly_session():
    return SessionLocal()
  
metadata = MetaData(
    naming_convention={
        "ck": "ck_%(table_name)s_%(constraint_name)s",
        "pk": "pk_%(table_name)s_%(column_0_name)s",
        "uq": "uq_%(table_name)s_%(column_0_name)s",
        "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
        "ix": "ix_%(table_name)s_%(column_0_name)s",
    }
)


class BaseModel(DeclarativeBase):
    metadata = metadata
    type_annotation_map = {datetime.datetime: DateTime(timezone=True)}

  
# models/api_key.py

class ApiKey(BaseModel):
  id: Mapped[intpk]
  key: Mapped[str] = mapped_column(unique=True)
  deactivated_at: Mapped[datetime | None]
  _deactivated_by: Mapped[int | None] = mapped_column(
      "deactivated_by", ForeignKey("accounts.id")
  )

  deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
  _deleted_by: Mapped[int | None] = mapped_column(
      "deleted_by", ForeignKey("accounts.id")
  )
    
  owner = relationship("Owner", back_populates="api_keys")
    
  @hybrid_property
  def is_active(self):  # type: ignore
      return self.deactivated_at is None and self.deleted_at is None

  @is_active.expression
  def is_active(cls):
      return cls.deactivated_at.is_(None) & cls.deleted_at.is_(None)  # type: ignore

# models/owner.py

class Owner(BaseModel):
  id: Mapped[intpk]
    
  active_items = relationship(
      "Item",
      secondary="items_owners",
      secondaryjoin=f"and_(Item.is_active.is_(True), Item.id == ItemOwners.item_id)",
      viewonly=True,
  )
  api_keys = relationship("ApiKey", back_populates="owner")
    
# deps/key_check.py

from fastapi import Depends
from sqlalchemy import select

from db import get_raw_readonly_session
from models.api_key import ApiKey
from models.owner import Owner

def lookup_api_key(session, key):
  return session.scalar(
    select(ApiKey)
    .options(joinedload(ApiKey.owner).joinedload(Owner.active_items))
    .where(ApiKey.is_active)
    .where(ApiKey.key = key)
  )

async def resolve_completion_context(
  api_key = Header(None),
  raw_session == Depends(get_raw_readonly_session)
):
  with raw_session as session:
    key = await trio.to_thread.run_sync(lookup_api_key)