internal.queries.token 

 1# Code generated by sqlc. DO NOT EDIT.
 2# versions:
 3#   sqlc v1.30.0
 4# source: token.sql
 5import datetime
 6import pydantic
 7from typing import Optional
 8
 9import sqlalchemy
10import sqlalchemy.ext.asyncio
11
12from internal.queries import models
13
14
15CREATE_TOKEN = """-- name: create_token \\:one
16INSERT INTO token (user_id, token, expires_at)
17VALUES (:p1, :p2, :p3)
18RETURNING token_id, user_id, token, created_at, expires_at
19"""
20
21
22class CreateTokenParams(pydantic.BaseModel):
23    user_id: int
24    token: str
25    expires_at: datetime.datetime
26
27
28DELETE_TOKEN = """-- name: delete_token \\:one
29DELETE FROM token
30WHERE token = :p1
31RETURNING token_id, user_id, token, created_at, expires_at
32"""
33
34
35GET_SESSION_BY_TOKEN = """-- name: get_session_by_token \\:one
36SELECT u.user_id, u.username, u.email, u.role, t.token, t.expires_at, t.created_at
37FROM token t
38JOIN users u ON u.user_id = t.user_id
39WHERE t.token = :p1
40AND t.expires_at > NOW()
41LIMIT 1
42"""
43
44
45class GetSessionByTokenRow(pydantic.BaseModel):
46    user_id: int
47    username: str
48    email: str
49    role: models.UserRole
50    token: str
51    expires_at: datetime.datetime
52    created_at: datetime.datetime
53
54
55class AsyncQuerier:
56    def __init__(self, conn: sqlalchemy.ext.asyncio.AsyncConnection):
57        self._conn = conn
58
59    async def create_token(self, arg: CreateTokenParams) -> Optional[models.Token]:
60        row = (await self._conn.execute(sqlalchemy.text(CREATE_TOKEN), {"p1": arg.user_id, "p2": arg.token, "p3": arg.expires_at})).first()
61        if row is None:
62            return None
63        return models.Token(
64            token_id=row[0],
65            user_id=row[1],
66            token=row[2],
67            created_at=row[3],
68            expires_at=row[4],
69        )
70
71    async def delete_token(self, *, token: str) -> Optional[models.Token]:
72        row = (await self._conn.execute(sqlalchemy.text(DELETE_TOKEN), {"p1": token})).first()
73        if row is None:
74            return None
75        return models.Token(
76            token_id=row[0],
77            user_id=row[1],
78            token=row[2],
79            created_at=row[3],
80            expires_at=row[4],
81        )
82
83    async def get_session_by_token(self, *, token: str) -> Optional[GetSessionByTokenRow]:
84        row = (await self._conn.execute(sqlalchemy.text(GET_SESSION_BY_TOKEN), {"p1": token})).first()
85        if row is None:
86            return None
87        return GetSessionByTokenRow(
88            user_id=row[0],
89            username=row[1],
90            email=row[2],
91            role=row[3],
92            token=row[4],
93            expires_at=row[5],
94            created_at=row[6],
95        )
CREATE_TOKEN = '-- name: create_token \\:one\nINSERT INTO token (user_id, token, expires_at)\nVALUES (:p1, :p2, :p3)\nRETURNING token_id, user_id, token, created_at, expires_at\n'
class CreateTokenParams(pydantic.main.BaseModel): 
23class CreateTokenParams(pydantic.BaseModel):
24    user_id: int
25    token: str
26    expires_at: datetime.datetime

!!! abstract "Usage Documentation" Models

A base class for creating Pydantic models.

Attributes:
  • __class_vars__: The names of the class variables defined on the model.
  • __private_attributes__: Metadata about the private attributes of the model.
  • __signature__: The synthesized __init__ [Signature][inspect.Signature] of the model.
  • __pydantic_complete__: Whether model building is completed, or if there are still undefined fields.
  • __pydantic_core_schema__: The core schema of the model.
  • __pydantic_custom_init__: Whether the model has a custom __init__ function.
  • __pydantic_decorators__: Metadata containing the decorators defined on the model. This replaces Model.__validators__ and Model.__root_validators__ from Pydantic V1.
  • __pydantic_generic_metadata__: Metadata for generic models; contains data used for a similar purpose to __args__, __origin__, __parameters__ in typing-module generics. May eventually be replaced by these.
  • __pydantic_parent_namespace__: Parent namespace of the model, used for automatic rebuilding of models.
  • __pydantic_post_init__: The name of the post-init method for the model, if defined.
  • __pydantic_root_model__: Whether the model is a [RootModel][pydantic.root_model.RootModel].
  • __pydantic_serializer__: The pydantic-core SchemaSerializer used to dump instances of the model.
  • __pydantic_validator__: The pydantic-core SchemaValidator used to validate instances of the model.
  • __pydantic_fields__: A dictionary of field names and their corresponding [FieldInfo][pydantic.fields.FieldInfo] objects.
  • __pydantic_computed_fields__: A dictionary of computed field names and their corresponding [ComputedFieldInfo][pydantic.fields.ComputedFieldInfo] objects.
  • __pydantic_extra__: A dictionary containing extra values, if [extra][pydantic.config.ConfigDict.extra] is set to 'allow'.
  • __pydantic_fields_set__: The names of fields explicitly set during instantiation.
  • __pydantic_private__: Values of private attributes set on the model instance.
user_id: int = PydanticUndefined
token: str = PydanticUndefined
expires_at: datetime.datetime = PydanticUndefined
DELETE_TOKEN = '-- name: delete_token \\:one\nDELETE FROM token\nWHERE token = :p1\nRETURNING token_id, user_id, token, created_at, expires_at\n'
GET_SESSION_BY_TOKEN = '-- name: get_session_by_token \\:one\nSELECT u.user_id, u.username, u.email, u.role, t.token, t.expires_at, t.created_at\nFROM token t\nJOIN users u ON u.user_id = t.user_id\nWHERE t.token = :p1\nAND t.expires_at > NOW()\nLIMIT 1\n'
class GetSessionByTokenRow(pydantic.main.BaseModel): 
46class GetSessionByTokenRow(pydantic.BaseModel):
47    user_id: int
48    username: str
49    email: str
50    role: models.UserRole
51    token: str
52    expires_at: datetime.datetime
53    created_at: datetime.datetime

!!! abstract "Usage Documentation" Models

A base class for creating Pydantic models.

Attributes:
  • __class_vars__: The names of the class variables defined on the model.
  • __private_attributes__: Metadata about the private attributes of the model.
  • __signature__: The synthesized __init__ [Signature][inspect.Signature] of the model.
  • __pydantic_complete__: Whether model building is completed, or if there are still undefined fields.
  • __pydantic_core_schema__: The core schema of the model.
  • __pydantic_custom_init__: Whether the model has a custom __init__ function.
  • __pydantic_decorators__: Metadata containing the decorators defined on the model. This replaces Model.__validators__ and Model.__root_validators__ from Pydantic V1.
  • __pydantic_generic_metadata__: Metadata for generic models; contains data used for a similar purpose to __args__, __origin__, __parameters__ in typing-module generics. May eventually be replaced by these.
  • __pydantic_parent_namespace__: Parent namespace of the model, used for automatic rebuilding of models.
  • __pydantic_post_init__: The name of the post-init method for the model, if defined.
  • __pydantic_root_model__: Whether the model is a [RootModel][pydantic.root_model.RootModel].
  • __pydantic_serializer__: The pydantic-core SchemaSerializer used to dump instances of the model.
  • __pydantic_validator__: The pydantic-core SchemaValidator used to validate instances of the model.
  • __pydantic_fields__: A dictionary of field names and their corresponding [FieldInfo][pydantic.fields.FieldInfo] objects.
  • __pydantic_computed_fields__: A dictionary of computed field names and their corresponding [ComputedFieldInfo][pydantic.fields.ComputedFieldInfo] objects.
  • __pydantic_extra__: A dictionary containing extra values, if [extra][pydantic.config.ConfigDict.extra] is set to 'allow'.
  • __pydantic_fields_set__: The names of fields explicitly set during instantiation.
  • __pydantic_private__: Values of private attributes set on the model instance.
user_id: int = PydanticUndefined
username: str = PydanticUndefined
email: str = PydanticUndefined
role: internal.queries.models.UserRole = PydanticUndefined
token: str = PydanticUndefined
expires_at: datetime.datetime = PydanticUndefined
created_at: datetime.datetime = PydanticUndefined
class AsyncQuerier: 
56class AsyncQuerier:
57    def __init__(self, conn: sqlalchemy.ext.asyncio.AsyncConnection):
58        self._conn = conn
59
60    async def create_token(self, arg: CreateTokenParams) -> Optional[models.Token]:
61        row = (await self._conn.execute(sqlalchemy.text(CREATE_TOKEN), {"p1": arg.user_id, "p2": arg.token, "p3": arg.expires_at})).first()
62        if row is None:
63            return None
64        return models.Token(
65            token_id=row[0],
66            user_id=row[1],
67            token=row[2],
68            created_at=row[3],
69            expires_at=row[4],
70        )
71
72    async def delete_token(self, *, token: str) -> Optional[models.Token]:
73        row = (await self._conn.execute(sqlalchemy.text(DELETE_TOKEN), {"p1": token})).first()
74        if row is None:
75            return None
76        return models.Token(
77            token_id=row[0],
78            user_id=row[1],
79            token=row[2],
80            created_at=row[3],
81            expires_at=row[4],
82        )
83
84    async def get_session_by_token(self, *, token: str) -> Optional[GetSessionByTokenRow]:
85        row = (await self._conn.execute(sqlalchemy.text(GET_SESSION_BY_TOKEN), {"p1": token})).first()
86        if row is None:
87            return None
88        return GetSessionByTokenRow(
89            user_id=row[0],
90            username=row[1],
91            email=row[2],
92            role=row[3],
93            token=row[4],
94            expires_at=row[5],
95            created_at=row[6],
96        )
AsyncQuerier(conn: sqlalchemy.ext.asyncio.engine.AsyncConnection) 
57    def __init__(self, conn: sqlalchemy.ext.asyncio.AsyncConnection):
58        self._conn = conn
async def create_token( self, arg: CreateTokenParams) -> internal.queries.models.Token | None: 
60    async def create_token(self, arg: CreateTokenParams) -> Optional[models.Token]:
61        row = (await self._conn.execute(sqlalchemy.text(CREATE_TOKEN), {"p1": arg.user_id, "p2": arg.token, "p3": arg.expires_at})).first()
62        if row is None:
63            return None
64        return models.Token(
65            token_id=row[0],
66            user_id=row[1],
67            token=row[2],
68            created_at=row[3],
69            expires_at=row[4],
70        )
async def delete_token(self, *, token: str) -> internal.queries.models.Token | None: 
72    async def delete_token(self, *, token: str) -> Optional[models.Token]:
73        row = (await self._conn.execute(sqlalchemy.text(DELETE_TOKEN), {"p1": token})).first()
74        if row is None:
75            return None
76        return models.Token(
77            token_id=row[0],
78            user_id=row[1],
79            token=row[2],
80            created_at=row[3],
81            expires_at=row[4],
82        )
async def get_session_by_token( self, *, token: str) -> GetSessionByTokenRow | None: 
84    async def get_session_by_token(self, *, token: str) -> Optional[GetSessionByTokenRow]:
85        row = (await self._conn.execute(sqlalchemy.text(GET_SESSION_BY_TOKEN), {"p1": token})).first()
86        if row is None:
87            return None
88        return GetSessionByTokenRow(
89            user_id=row[0],
90            username=row[1],
91            email=row[2],
92            role=row[3],
93            token=row[4],
94            expires_at=row[5],
95            created_at=row[6],
96        )