internal.queries.badge

  1# Code generated by sqlc. DO NOT EDIT.
  2# versions:
  3#   sqlc v1.30.0
  4# source: badge.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
 15ACQUIRE_BADGE = """-- name: acquire_badge \\:one
 16INSERT INTO badges_acquired (user_id, badge_id, level)
 17VALUES (:p1, :p2, :p3)
 18RETURNING user_id, badge_id, level, acquired_at
 19"""
 20
 21
 22class AcquireBadgeParams(pydantic.BaseModel):
 23    user_id: int
 24    badge_id: int
 25    level: int
 26
 27
 28GET_BADGE = """-- name: get_badge \\:one
 29SELECT badge_id, name, description
 30FROM badges
 31WHERE badge_id = :p1
 32LIMIT 1
 33"""
 34
 35
 36GET_BADGES = """-- name: get_badges \\:many
 37SELECT badge_id, name, description
 38FROM badges
 39"""
 40
 41
 42GET_CONSUMER_BADGES = """-- name: get_consumer_badges \\:many
 43SELECT b.badge_id, b.name, b.description, ba.level, ba.acquired_at
 44FROM badges b
 45INNER JOIN badges_acquired ba ON ba.badge_id = b.badge_id
 46WHERE ba.user_id = :p1
 47"""
 48
 49
 50class GetConsumerBadgesRow(pydantic.BaseModel):
 51    badge_id: int
 52    name: str
 53    description: str
 54    level: int
 55    acquired_at: datetime.datetime
 56
 57
 58UPDATE_BADGE = """-- name: update_badge \\:one
 59UPDATE badges
 60SET name = :p2, description = :p3
 61WHERE badge_id = :p1
 62RETURNING badge_id, name, description
 63"""
 64
 65
 66class UpdateBadgeParams(pydantic.BaseModel):
 67    badge_id: int
 68    name: str
 69    description: str
 70
 71
 72UPDATE_BADGE_LEVEL = """-- name: update_badge_level \\:one
 73UPDATE badges_acquired
 74SET level=:p1
 75WHERE user_id=:p2 AND badge_id=:p3
 76RETURNING user_id, badge_id, level, acquired_at
 77"""
 78
 79
 80class UpdateBadgeLevelParams(pydantic.BaseModel):
 81    level: int
 82    user_id: int
 83    badge_id: int
 84
 85
 86class AsyncQuerier:
 87    def __init__(self, conn: sqlalchemy.ext.asyncio.AsyncConnection):
 88        self._conn = conn
 89
 90    async def acquire_badge(self, arg: AcquireBadgeParams) -> Optional[models.BadgesAcquired]:
 91        row = (await self._conn.execute(sqlalchemy.text(ACQUIRE_BADGE), {"p1": arg.user_id, "p2": arg.badge_id, "p3": arg.level})).first()
 92        if row is None:
 93            return None
 94        return models.BadgesAcquired(
 95            user_id=row[0],
 96            badge_id=row[1],
 97            level=row[2],
 98            acquired_at=row[3],
 99        )
100
101    async def get_badge(self, *, badge_id: int) -> Optional[models.Badge]:
102        row = (await self._conn.execute(sqlalchemy.text(GET_BADGE), {"p1": badge_id})).first()
103        if row is None:
104            return None
105        return models.Badge(
106            badge_id=row[0],
107            name=row[1],
108            description=row[2],
109        )
110
111    async def get_badges(self) -> AsyncIterator[models.Badge]:
112        result = await self._conn.stream(sqlalchemy.text(GET_BADGES))
113        async for row in result:
114            yield models.Badge(
115                badge_id=row[0],
116                name=row[1],
117                description=row[2],
118            )
119
120    async def get_consumer_badges(self, *, user_id: int) -> AsyncIterator[GetConsumerBadgesRow]:
121        result = await self._conn.stream(sqlalchemy.text(GET_CONSUMER_BADGES), {"p1": user_id})
122        async for row in result:
123            yield GetConsumerBadgesRow(
124                badge_id=row[0],
125                name=row[1],
126                description=row[2],
127                level=row[3],
128                acquired_at=row[4],
129            )
130
131    async def update_badge(self, arg: UpdateBadgeParams) -> Optional[models.Badge]:
132        row = (await self._conn.execute(sqlalchemy.text(UPDATE_BADGE), {"p1": arg.badge_id, "p2": arg.name, "p3": arg.description})).first()
133        if row is None:
134            return None
135        return models.Badge(
136            badge_id=row[0],
137            name=row[1],
138            description=row[2],
139        )
140
141    async def update_badge_level(self, arg: UpdateBadgeLevelParams) -> Optional[models.BadgesAcquired]:
142        row = (await self._conn.execute(sqlalchemy.text(UPDATE_BADGE_LEVEL), {"p1": arg.level, "p2": arg.user_id, "p3": arg.badge_id})).first()
143        if row is None:
144            return None
145        return models.BadgesAcquired(
146            user_id=row[0],
147            badge_id=row[1],
148            level=row[2],
149            acquired_at=row[3],
150        )