internal.queries.consumer

  1# Code generated by sqlc. DO NOT EDIT.
  2# versions:
  3#   sqlc v1.30.0
  4# source: consumer.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_CONSUMER = """-- name: create_consumer \\:one
 16INSERT INTO consumers (user_id, fName, lName)
 17VALUES (:p1, :p2, :p3)
 18RETURNING user_id, fname, lname
 19"""
 20
 21
 22class CreateConsumerParams(pydantic.BaseModel):
 23    user_id: int
 24    fname: str
 25    lname: str
 26
 27
 28GET_CONSUMER = """-- name: get_consumer \\:one
 29SELECT u.user_id, u.username, u.email, c.fName, c.lName, u.last_login, u.created_at
 30FROM consumers  c
 31INNER JOIN users u ON c.user_id = u.user_id
 32WHERE u.user_id=:p1
 33LIMIT 1
 34"""
 35
 36
 37class GetConsumerRow(pydantic.BaseModel):
 38    user_id: int
 39    username: str
 40    email: str
 41    fname: str
 42    lname: str
 43    last_login: datetime.datetime
 44    created_at: datetime.datetime
 45
 46
 47GET_CONSUMERS = """-- name: get_consumers \\:many
 48SELECT u.user_id, u.username, u.email, c.fName, c.lName, u.last_login, u.created_at
 49FROM consumers c
 50INNER JOIN users u ON c.user_id = u.user_id
 51"""
 52
 53
 54class GetConsumersRow(pydantic.BaseModel):
 55    user_id: int
 56    username: str
 57    email: str
 58    fname: str
 59    lname: str
 60    last_login: datetime.datetime
 61    created_at: datetime.datetime
 62
 63
 64UPDATE_CONSUMER = """-- name: update_consumer \\:one
 65UPDATE consumers
 66SET fName=:p2, lName=:p3
 67WHERE user_id=:p1
 68RETURNING user_id, fname, lname
 69"""
 70
 71
 72class UpdateConsumerParams(pydantic.BaseModel):
 73    user_id: int
 74    fname: str
 75    lname: str
 76
 77
 78class AsyncQuerier:
 79    def __init__(self, conn: sqlalchemy.ext.asyncio.AsyncConnection):
 80        self._conn = conn
 81
 82    async def create_consumer(self, arg: CreateConsumerParams) -> Optional[models.Consumer]:
 83        row = (await self._conn.execute(sqlalchemy.text(CREATE_CONSUMER), {"p1": arg.user_id, "p2": arg.fname, "p3": arg.lname})).first()
 84        if row is None:
 85            return None
 86        return models.Consumer(
 87            user_id=row[0],
 88            fname=row[1],
 89            lname=row[2],
 90        )
 91
 92    async def get_consumer(self, *, user_id: int) -> Optional[GetConsumerRow]:
 93        row = (await self._conn.execute(sqlalchemy.text(GET_CONSUMER), {"p1": user_id})).first()
 94        if row is None:
 95            return None
 96        return GetConsumerRow(
 97            user_id=row[0],
 98            username=row[1],
 99            email=row[2],
100            fname=row[3],
101            lname=row[4],
102            last_login=row[5],
103            created_at=row[6],
104        )
105
106    async def get_consumers(self) -> AsyncIterator[GetConsumersRow]:
107        result = await self._conn.stream(sqlalchemy.text(GET_CONSUMERS))
108        async for row in result:
109            yield GetConsumersRow(
110                user_id=row[0],
111                username=row[1],
112                email=row[2],
113                fname=row[3],
114                lname=row[4],
115                last_login=row[5],
116                created_at=row[6],
117            )
118
119    async def update_consumer(self, arg: UpdateConsumerParams) -> Optional[models.Consumer]:
120        row = (await self._conn.execute(sqlalchemy.text(UPDATE_CONSUMER), {"p1": arg.user_id, "p2": arg.fname, "p3": arg.lname})).first()
121        if row is None:
122            return None
123        return models.Consumer(
124            user_id=row[0],
125            fname=row[1],
126            lname=row[2],
127        )