Class tigrbl.engine.builders.HybridSession

tigrbl.engine.builders.HybridSession

Bases: AsyncSession

An AsyncSession that ALSO behaves like a synchronous Session for the handful of blocking helpers Tigrbl’s CRUD cores expect (query, commit, flush, refresh, get, delete).

query

query(*e, **k)

Source code in tigrbl/engine/builders.py

def query(self, *e, **k):
    logger.debug("HybridSession.query called with args=%r kwargs=%r", e, k)
    return self.sync_session.query(*e, **k)

add

add(*a, **k)

Source code in tigrbl/engine/builders.py

def add(self, *a, **k):
    logger.debug("HybridSession.add called with args=%r kwargs=%r", a, k)
    return self.sync_session.add(*a, **k)

get async

get(*a, **k)

Source code in tigrbl/engine/builders.py

async def get(self, *a, **k):
    logger.debug("HybridSession.get called with args=%r kwargs=%r", a, k)
    return await super().get(*a, **k)

flush async

flush(*a, **k)

Source code in tigrbl/engine/builders.py

async def flush(self, *a, **k):
    logger.debug("HybridSession.flush called with args=%r kwargs=%r", a, k)
    return await super().flush(*a, **k)

commit async

commit(*a, **k)

Source code in tigrbl/engine/builders.py

async def commit(self, *a, **k):
    logger.debug("HybridSession.commit called with args=%r kwargs=%r", a, k)
    return await super().commit(*a, **k)

refresh async

refresh(*a, **k)

Source code in tigrbl/engine/builders.py

async def refresh(self, *a, **k):
    logger.debug("HybridSession.refresh called with args=%r kwargs=%r", a, k)
    return await super().refresh(*a, **k)

delete async

delete(*a, **k)

Source code in tigrbl/engine/builders.py

async def delete(self, *a, **k):
    logger.debug("HybridSession.delete called with args=%r kwargs=%r", a, k)
    return await super().delete(*a, **k)

run_sync async

run_sync(fn, *a, **kw)

Source code in tigrbl/engine/builders.py

async def run_sync(self, fn, *a, **kw):
    logger.debug(
        "HybridSession.run_sync called with fn=%r args=%r kwargs=%r", fn, a, kw
    )
    try:
        rv = await super().run_sync(fn, *a, **kw)
        logger.debug("HybridSession.run_sync succeeded with result=%r", rv)
        return rv
    except (OSError, SQLAlchemyError) as exc:
        url = getattr(self.bind, "url", "unknown")
        logger.debug(
            "HybridSession.run_sync failed for url %s with exc=%r", url, exc
        )
        await self.bind.dispose()
        raise RuntimeError(
            f"Failed to connect to database at '{url}'. "
            "Ensure the database is reachable and credentials are correct."
        ) from exc