async def METHOD_NAME(
    session: sa.orm.Session,
    tag: str,
    db: PrefectDBInterface,
):
    """
    Reads a concurrency limit by tag. If used for orchestration, simultaneous read race
    conditions might allow the concurrency limit to be temporarily exceeded.
    """
    query = sa.select(db.ConcurrencyLimit).where(db.ConcurrencyLimit.tag == tag)
    result = await session.execute(query)
    return result.scalar()