internal.queries.user

  1# Code generated by sqlc. DO NOT EDIT.
  2# versions:
  3#   sqlc v1.30.0
  4# source: user.sql
  5import datetime
  6import decimal
  7import pydantic
  8from typing import AsyncIterator, Optional
  9
 10import sqlalchemy
 11import sqlalchemy.ext.asyncio
 12
 13from internal.queries import models
 14
 15
 16CREATE_USER = """-- name: create_user \\:one
 17INSERT INTO users (email, username, pw_hash, role)
 18VALUES (:p1, :p2, :p3, :p4)
 19RETURNING user_id, username, email, role, created_at
 20"""
 21
 22
 23class CreateUserParams(pydantic.BaseModel):
 24    email: str
 25    username: str
 26    pw_hash: str
 27    role: models.UserRole
 28
 29
 30class CreateUserRow(pydantic.BaseModel):
 31    user_id: int
 32    username: str
 33    email: str
 34    role: models.UserRole
 35    created_at: datetime.datetime
 36
 37
 38DELETE_USER = """-- name: delete_user \\:one
 39DELETE FROM users
 40WHERE user_id = :p1
 41RETURNING user_id, username, email, role, created_at
 42"""
 43
 44
 45class DeleteUserRow(pydantic.BaseModel):
 46    user_id: int
 47    username: str
 48    email: str
 49    role: models.UserRole
 50    created_at: datetime.datetime
 51
 52
 53GET_USER = """-- name: get_user \\:one
 54SELECT user_id, username, email, role, created_at
 55FROM users
 56WHERE user_id = :p1 
 57LIMIT 1
 58"""
 59
 60
 61class GetUserRow(pydantic.BaseModel):
 62    user_id: int
 63    username: str
 64    email: str
 65    role: models.UserRole
 66    created_at: datetime.datetime
 67
 68
 69GET_USER_ID = """-- name: get_user_id \\:one
 70SELECT user_id, role
 71FROM users
 72WHERE username=:p1
 73LIMIT 1
 74"""
 75
 76
 77class GetUserIdRow(pydantic.BaseModel):
 78    user_id: int
 79    role: models.UserRole
 80
 81
 82GET_USER_LOGIN = """-- name: get_user_login \\:one
 83SELECT user_id, username, email, pw_hash, role
 84FROM users
 85WHERE email = :p1
 86LIMIT 1
 87"""
 88
 89
 90class GetUserLoginRow(pydantic.BaseModel):
 91    user_id: int
 92    username: str
 93    email: str
 94    pw_hash: str
 95    role: models.UserRole
 96
 97
 98GET_USERS = """-- name: get_users \\:many
 99SELECT user_id, username, email, role, created_at, last_login
100FROM users
101"""
102
103
104class GetUsersRow(pydantic.BaseModel):
105    user_id: int
106    username: str
107    email: str
108    role: models.UserRole
109    created_at: datetime.datetime
110    last_login: datetime.datetime
111
112
113LEADERBOARD_CARBON_DIOXIDE = """-- name: leaderboard_carbon_dioxide \\:many
114SELECT u.username, COALESCE(SUM(b.carbon_dioxide), 0)\\:\\:numeric AS total_carbon_dioxide
115FROM users u
116LEFT JOIN reservations r ON r.consumer_id = u.user_id
117LEFT JOIN bundles b ON b.bundle_id = r.bundle_id
118GROUP BY u.user_id, u.username
119ORDER BY total_carbon_dioxide DESC
120LIMIT :p1
121"""
122
123
124class LeaderboardCarbonDioxideRow(pydantic.BaseModel):
125    username: str
126    total_carbon_dioxide: decimal.Decimal
127
128
129LEADERBOARD_RESERVATIONS = """-- name: leaderboard_reservations \\:many
130SELECT u.username, COUNT(r.reservation_id) AS reservation_count
131FROM users u
132LEFT JOIN reservations r ON r.consumer_id = u.user_id
133GROUP BY u.user_id, u.username
134ORDER BY reservation_count DESC
135LIMIT :p1
136"""
137
138
139class LeaderboardReservationsRow(pydantic.BaseModel):
140    username: str
141    reservation_count: int
142
143
144UPDATE_USER_EMAIL = """-- name: update_user_email \\:one
145UPDATE users
146SET email=:p2
147WHERE user_id=:p1
148RETURNING user_id, username, email, role, created_at
149"""
150
151
152class UpdateUserEmailParams(pydantic.BaseModel):
153    user_id: int
154    email: str
155
156
157class UpdateUserEmailRow(pydantic.BaseModel):
158    user_id: int
159    username: str
160    email: str
161    role: models.UserRole
162    created_at: datetime.datetime
163
164
165UPDATE_USER_PASSWORD = """-- name: update_user_password \\:one
166UPDATE users
167SET pw_hash=:p2
168WHERE user_id=:p1
169RETURNING user_id, username, email, role, created_at
170"""
171
172
173class UpdateUserPasswordParams(pydantic.BaseModel):
174    user_id: int
175    pw_hash: str
176
177
178class UpdateUserPasswordRow(pydantic.BaseModel):
179    user_id: int
180    username: str
181    email: str
182    role: models.UserRole
183    created_at: datetime.datetime
184
185
186class AsyncQuerier:
187    def __init__(self, conn: sqlalchemy.ext.asyncio.AsyncConnection):
188        self._conn = conn
189
190    async def create_user(self, arg: CreateUserParams) -> Optional[CreateUserRow]:
191        row = (await self._conn.execute(sqlalchemy.text(CREATE_USER), {
192            "p1": arg.email,
193            "p2": arg.username,
194            "p3": arg.pw_hash,
195            "p4": arg.role,
196        })).first()
197        if row is None:
198            return None
199        return CreateUserRow(
200            user_id=row[0],
201            username=row[1],
202            email=row[2],
203            role=row[3],
204            created_at=row[4],
205        )
206
207    async def delete_user(self, *, user_id: int) -> Optional[DeleteUserRow]:
208        row = (await self._conn.execute(sqlalchemy.text(DELETE_USER), {"p1": user_id})).first()
209        if row is None:
210            return None
211        return DeleteUserRow(
212            user_id=row[0],
213            username=row[1],
214            email=row[2],
215            role=row[3],
216            created_at=row[4],
217        )
218
219    async def get_user(self, *, user_id: int) -> Optional[GetUserRow]:
220        row = (await self._conn.execute(sqlalchemy.text(GET_USER), {"p1": user_id})).first()
221        if row is None:
222            return None
223        return GetUserRow(
224            user_id=row[0],
225            username=row[1],
226            email=row[2],
227            role=row[3],
228            created_at=row[4],
229        )
230
231    async def get_user_id(self, *, username: str) -> Optional[GetUserIdRow]:
232        row = (await self._conn.execute(sqlalchemy.text(GET_USER_ID), {"p1": username})).first()
233        if row is None:
234            return None
235        return GetUserIdRow(
236            user_id=row[0],
237            role=row[1],
238        )
239
240    async def get_user_login(self, *, email: str) -> Optional[GetUserLoginRow]:
241        row = (await self._conn.execute(sqlalchemy.text(GET_USER_LOGIN), {"p1": email})).first()
242        if row is None:
243            return None
244        return GetUserLoginRow(
245            user_id=row[0],
246            username=row[1],
247            email=row[2],
248            pw_hash=row[3],
249            role=row[4],
250        )
251
252    async def get_users(self) -> AsyncIterator[GetUsersRow]:
253        result = await self._conn.stream(sqlalchemy.text(GET_USERS))
254        async for row in result:
255            yield GetUsersRow(
256                user_id=row[0],
257                username=row[1],
258                email=row[2],
259                role=row[3],
260                created_at=row[4],
261                last_login=row[5],
262            )
263
264    async def leaderboard_carbon_dioxide(self, *, limit: int) -> AsyncIterator[LeaderboardCarbonDioxideRow]:
265        result = await self._conn.stream(sqlalchemy.text(LEADERBOARD_CARBON_DIOXIDE), {"p1": limit})
266        async for row in result:
267            yield LeaderboardCarbonDioxideRow(
268                username=row[0],
269                total_carbon_dioxide=row[1],
270            )
271
272    async def leaderboard_reservations(self, *, limit: int) -> AsyncIterator[LeaderboardReservationsRow]:
273        result = await self._conn.stream(sqlalchemy.text(LEADERBOARD_RESERVATIONS), {"p1": limit})
274        async for row in result:
275            yield LeaderboardReservationsRow(
276                username=row[0],
277                reservation_count=row[1],
278            )
279
280    async def update_user_email(self, arg: UpdateUserEmailParams) -> Optional[UpdateUserEmailRow]:
281        row = (await self._conn.execute(sqlalchemy.text(UPDATE_USER_EMAIL), {"p1": arg.user_id, "p2": arg.email})).first()
282        if row is None:
283            return None
284        return UpdateUserEmailRow(
285            user_id=row[0],
286            username=row[1],
287            email=row[2],
288            role=row[3],
289            created_at=row[4],
290        )
291
292    async def update_user_password(self, arg: UpdateUserPasswordParams) -> Optional[UpdateUserPasswordRow]:
293        row = (await self._conn.execute(sqlalchemy.text(UPDATE_USER_PASSWORD), {"p1": arg.user_id, "p2": arg.pw_hash})).first()
294        if row is None:
295            return None
296        return UpdateUserPasswordRow(
297            user_id=row[0],
298            username=row[1],
299            email=row[2],
300            role=row[3],
301            created_at=row[4],
302        )