internal.queries.admin

  1# Code generated by sqlc. DO NOT EDIT.
  2# versions:
  3#   sqlc v1.30.0
  4# source: admin.sql
  5import datetime
  6import pydantic
  7from typing import AsyncIterator, Optional
  8
  9import sqlalchemy
 10import sqlalchemy.ext.asyncio
 11
 12from internal.queries import models
 13
 14
 15CREATE_ADMIN = """-- name: create_admin \\:one
 16INSERT INTO admins (user_id, fName, lName)
 17VALUES (:p1,:p2,:p3)
 18RETURNING user_id, fname, lname, active
 19"""
 20
 21
 22class CreateAdminParams(pydantic.BaseModel):
 23    user_id: int
 24    fname: str
 25    lname: str
 26
 27
 28GET_ADMIN = """-- name: get_admin \\:one
 29SELECT u.user_id, u.username, u.email, a.fName, a.lName, a.active, u.last_login, u.created_at
 30FROM admins a
 31INNER JOIN users u ON a.user_id = u.user_id
 32WHERE u.user_id = :p1
 33LIMIT 1
 34"""
 35
 36
 37class GetAdminRow(pydantic.BaseModel):
 38    user_id: int
 39    username: str
 40    email: str
 41    fname: str
 42    lname: str
 43    active: Optional[bool]
 44    last_login: datetime.datetime
 45    created_at: datetime.datetime
 46
 47
 48GET_ADMINS = """-- name: get_admins \\:many
 49SELECT u.user_id, u.username, u.email, a.fName, a.lName, a.active, u.last_login, u.created_at
 50FROM admins a
 51INNER JOIN users u ON a.user_id = u.user_id
 52"""
 53
 54
 55class GetAdminsRow(pydantic.BaseModel):
 56    user_id: int
 57    username: str
 58    email: str
 59    fname: str
 60    lname: str
 61    active: Optional[bool]
 62    last_login: datetime.datetime
 63    created_at: datetime.datetime
 64
 65
 66SET_IS_ADMIN_ACTIVE = """-- name: set_is_admin_active \\:one
 67UPDATE admins
 68SET active=:p2
 69WHERE user_id=:p1
 70RETURNING user_id, fname, lname, active
 71"""
 72
 73
 74class SetIsAdminActiveParams(pydantic.BaseModel):
 75    user_id: int
 76    active: Optional[bool]
 77
 78
 79UPDATE_ADMIN = """-- name: update_admin \\:one
 80UPDATE admins
 81SET fName = :p2, lName = :p3
 82WHERE user_id = :p1
 83RETURNING user_id, fname, lname, active
 84"""
 85
 86
 87class UpdateAdminParams(pydantic.BaseModel):
 88    user_id: int
 89    fname: str
 90    lname: str
 91
 92
 93class AsyncQuerier:
 94    def __init__(self, conn: sqlalchemy.ext.asyncio.AsyncConnection):
 95        self._conn = conn
 96
 97    async def create_admin(self, arg: CreateAdminParams) -> Optional[models.Admin]:
 98        row = (await self._conn.execute(sqlalchemy.text(CREATE_ADMIN), {"p1": arg.user_id, "p2": arg.fname, "p3": arg.lname})).first()
 99        if row is None:
100            return None
101        return models.Admin(
102            user_id=row[0],
103            fname=row[1],
104            lname=row[2],
105            active=row[3],
106        )
107
108    async def get_admin(self, *, user_id: int) -> Optional[GetAdminRow]:
109        row = (await self._conn.execute(sqlalchemy.text(GET_ADMIN), {"p1": user_id})).first()
110        if row is None:
111            return None
112        return GetAdminRow(
113            user_id=row[0],
114            username=row[1],
115            email=row[2],
116            fname=row[3],
117            lname=row[4],
118            active=row[5],
119            last_login=row[6],
120            created_at=row[7],
121        )
122
123    async def get_admins(self) -> AsyncIterator[GetAdminsRow]:
124        result = await self._conn.stream(sqlalchemy.text(GET_ADMINS))
125        async for row in result:
126            yield GetAdminsRow(
127                user_id=row[0],
128                username=row[1],
129                email=row[2],
130                fname=row[3],
131                lname=row[4],
132                active=row[5],
133                last_login=row[6],
134                created_at=row[7],
135            )
136
137    async def set_is_admin_active(self, arg: SetIsAdminActiveParams) -> Optional[models.Admin]:
138        row = (await self._conn.execute(sqlalchemy.text(SET_IS_ADMIN_ACTIVE), {"p1": arg.user_id, "p2": arg.active})).first()
139        if row is None:
140            return None
141        return models.Admin(
142            user_id=row[0],
143            fname=row[1],
144            lname=row[2],
145            active=row[3],
146        )
147
148    async def update_admin(self, arg: UpdateAdminParams) -> Optional[models.Admin]:
149        row = (await self._conn.execute(sqlalchemy.text(UPDATE_ADMIN), {"p1": arg.user_id, "p2": arg.fname, "p3": arg.lname})).first()
150        if row is None:
151            return None
152        return models.Admin(
153            user_id=row[0],
154            fname=row[1],
155            lname=row[2],
156            active=row[3],
157        )