Skip to content

Class tigrbl_auth.orm.auth_session.AuthSession

tigrbl_auth.orm.auth_session.AuthSession

Bases: Base, GUIDPk, Timestamped, UserColumn, TenantColumn

username class-attribute instance-attribute 

username = acol(storage=S(String(120), nullable=False))

auth_time class-attribute instance-attribute 

auth_time = acol(
    storage=S(
        TZDateTime, nullable=False, default=lambda: now(utc)
    )
)

created_at class-attribute instance-attribute 

created_at = acol(
    spec=ColumnSpec(
        storage=S(
            type_=TZDateTime,
            default=tzutcnow,
            nullable=False,
        ),
        field=F(py_type=datetime),
        io=RO_IO,
    )
)

updated_at class-attribute instance-attribute 

updated_at = acol(
    spec=ColumnSpec(
        storage=S(
            type_=TZDateTime,
            default=tzutcnow,
            onupdate=tzutcnow,
            nullable=False,
        ),
        field=F(py_type=datetime),
        io=RO_IO,
    )
)

id class-attribute instance-attribute 

id = acol(
    spec=ColumnSpec(
        storage=S(
            type_=PgUUID(as_uuid=True),
            primary_key=True,
            default=uuid4,
        ),
        field=F(
            py_type=UUID,
            constraints={"examples": [uuid_example]},
        ),
        io=RO_IO,
    )
)

metadata class-attribute instance-attribute 

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

login async 

login(ctx)
Source code in tigrbl_auth/orm/auth_session.py 
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
@op_ctx(
    alias="login",
    target="create",
    arity="collection",
    request_schema=CredsIn,
    response_schema=TokenPair,
)
async def login(cls, ctx):
    import secrets
    from ..rfc.rfc8414_metadata import ISSUER
    from ..oidc_id_token import mint_id_token
    from ..routers.shared import _jwt, _require_tls

    request = ctx.get("request")
    _require_tls(request)
    payload = ctx.get("payload") or {}
    session = await cls.handlers.create.core(
        {"payload": payload, "db": ctx.get("db")}
    )
    access, refresh = await _jwt.async_sign_pair(
        sub=str(session.user_id),
        tid=str(session.tenant_id),
        scope="openid profile email",
    )
    id_token = await mint_id_token(
        sub=str(session.user_id),
        aud=ISSUER,
        nonce=secrets.token_urlsafe(8),
        issuer=ISSUER,
        sid=str(session.id),
    )
    pair = {
        "access_token": access,
        "refresh_token": refresh,
        "id_token": id_token,
    }
    response = JSONResponse(pair)
    response.set_cookie("sid", str(session.id), httponly=True, samesite="lax")
    return response

logout async 

logout(ctx)
Source code in tigrbl_auth/orm/auth_session.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
@op_ctx(
    alias="logout",
    target="delete",
    arity="collection",
    request_schema=LogoutIn,
    status_code=status.HTTP_204_NO_CONTENT,
)
async def logout(cls, ctx):
    from ..rfc.rfc8414_metadata import ISSUER
    from ..oidc_id_token import verify_id_token
    from ..routers.shared import (
        _require_tls,
        _front_channel_logout,
        _back_channel_logout,
    )

    request = ctx.get("request")
    _require_tls(request)
    payload = ctx.get("payload") or {}
    id_hint = payload.get("id_token_hint")
    try:
        claims = await verify_id_token(id_hint, issuer=ISSUER, audience=ISSUER)
    except Exception as exc:  # pragma: no cover - passthrough
        raise HTTPException(
            status.HTTP_400_BAD_REQUEST, "invalid id_token_hint"
        ) from exc
    sid = claims.get("sid")
    if sid:
        session = await cls.handlers.read.core({"payload": {"id": UUID(sid)}})
        if session:
            await cls.handlers.delete.core({"obj": session})
        await _front_channel_logout(sid)
        await _back_channel_logout(sid)
    response = Response(status_code=status.HTTP_204_NO_CONTENT)
    response.delete_cookie("sid")
    return response

tenant_id 

tenant_id()
Source code in tigrbl/orm/mixins/principals.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
@declared_attr
def tenant_id(cls) -> Mapped[UUID]:
    schema = getattr(cls, "__tenant_table_schema__", None) or _infer_schema(cls)
    spec = ColumnSpec(
        storage=S(
            type_=PgUUID(as_uuid=True),
            fk=ForeignKeySpec(target=f"{schema}.tenants.id"),
            nullable=False,
            index=True,
        ),
        field=F(py_type=UUID, constraints={"examples": [uuid_example]}),
        io=CRUD_IO,
    )
    return acol(spec=spec)

user_id 

user_id()
Source code in tigrbl/orm/mixins/principals.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
@declared_attr
def user_id(cls) -> Mapped[UUID]:
    schema = getattr(cls, "__user_table_schema__", None) or _infer_schema(cls)
    spec = ColumnSpec(
        storage=S(
            type_=PgUUID(as_uuid=True),
            fk=ForeignKeySpec(target=f"{schema}.users.id"),
            nullable=False,
            index=True,
        ),
        field=F(py_type=UUID, constraints={"examples": [uuid_example]}),
        io=CRUD_IO,
    )
    return acol(spec=spec)