Skip to content

Instantly share code, notes, and snippets.

@wonderbeyond

wonderbeyond/sa-1.4-adb.py

Last active June 13, 2023 06:03
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save wonderbeyond/122cb2a80e39475dd85a27772109e728 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save wonderbeyond/122cb2a80e39475dd85a27772109e728 to your computer and use it in GitHub Desktop.
Download ZIP

Revisions

  1. wonderbeyond revised this gist Jun 13, 2023. 1 changed file with 0 additions and 2 deletions.
    2 changes: 0 additions & 2 deletions sa-1.4-adb.py
    View file
    Original file line number Diff line number Diff line change
    @@ -18,8 +18,6 @@ def __init__(
    session_options.setdefault("bind", self.engine)

    self.Session: Callable[..., AsyncSession] = sessionmaker(**session_options)
    session_options.pop("class_")
    self._session = AsyncSession(**session_options)

    def __repr__(self) -> str:
    return f"<AsyncSQLAlchemy('{self.url}')>"
  2. wonderbeyond revised this gist Jun 13, 2023. 1 changed file with 4 additions and 0 deletions.
    4 changes: 4 additions & 0 deletions sa-1.4-adb.py
    View file
    Original file line number Diff line number Diff line change
    @@ -36,3 +36,7 @@ def __repr__(self) -> str:
    "expire_on_commit": False,
    }
    )

    async with adb.Session() as s:
    user = (await s.execute(sa.select(User).limit(1))).scalar()
    print(user)
  3. wonderbeyond created this gist Jun 13, 2023.
    38 changes: 38 additions & 0 deletions sa-1.4-adb.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,38 @@
    class AsyncSQLAlchemy:
    def __init__(
    self,
    url: str,
    engine_options: Optional[Dict[str, Any]] = None,
    session_options: Optional[Dict[str, Any]] = None,
    ):
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy.ext.asyncio import AsyncSession

    self.url = url

    engine_options = engine_options or {}
    self.engine = create_async_engine(self.url, **engine_options)

    session_options = session_options or {}
    session_options.setdefault("class_", AsyncSession)
    session_options.setdefault("bind", self.engine)

    self.Session: Callable[..., AsyncSession] = sessionmaker(**session_options)
    session_options.pop("class_")
    self._session = AsyncSession(**session_options)

    def __repr__(self) -> str:
    return f"<AsyncSQLAlchemy('{self.url}')>"


    adb = AsyncSQLAlchemy(
    f"postgresql+asyncpg://{dbc.user}:{dbc.password}@{dbc.host}:{dbc.port}/{dbc.database}",
    engine_options={
    "echo": dbc.echo_sql,
    'pool_size': dbc.pool_size,
    'max_overflow': dbc.max_overflow,
    },
    session_options={
    "expire_on_commit": False,
    }
    )