Last active
April 20, 2023 19:41
-
-
Save ejlangev/c6b2917e16e968e04d949d74ffb224e7 to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| # 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) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment