internal.queries.category

  1# Code generated by sqlc. DO NOT EDIT.
  2# versions:
  3#   sqlc v1.30.0
  4# source: category.sql
  5import pydantic
  6from typing import AsyncIterator, Optional
  7
  8import sqlalchemy
  9import sqlalchemy.ext.asyncio
 10
 11from internal.queries import models
 12
 13
 14ADD_BUNDLES_CATEGORY = """-- name: add_bundles_category \\:one
 15INSERT INTO bundle_category (bundle_id, category_id)
 16VALUES (:p1,:p2)
 17RETURNING category_id, bundle_id
 18"""
 19
 20
 21class AddBundlesCategoryParams(pydantic.BaseModel):
 22    bundle_id: int
 23    category_id: int
 24
 25
 26CREATE_CATEGORY = """-- name: create_category \\:one
 27INSERT INTO category (category_name, category_coefficient)
 28VALUES (:p1, :p2)
 29RETURNING category_id, category_name, category_coefficient
 30"""
 31
 32
 33class CreateCategoryParams(pydantic.BaseModel):
 34    category_name: str
 35    category_coefficient: float
 36
 37
 38DELETE_BUNDLE_CATEGORY = """-- name: delete_bundle_category \\:one
 39DELETE FROM bundle_category
 40WHERE category_id=:p1 AND bundle_id=:p2
 41RETURNING category_id, bundle_id
 42"""
 43
 44
 45class DeleteBundleCategoryParams(pydantic.BaseModel):
 46    category_id: int
 47    bundle_id: int
 48
 49
 50DELETE_CATEGORY = """-- name: delete_category \\:one
 51DELETE FROM category
 52WHERE category_id = :p1
 53RETURNING category_id, category_name, category_coefficient
 54"""
 55
 56
 57GET_BUNDLE_CATEGORIES = """-- name: get_bundle_categories \\:many
 58SELECT c.category_id
 59FROM category c
 60JOIN bundle_category bc ON bc.category_id = c.category_id
 61JOIN bundles b ON b.bundle_id = bc.bundle_id
 62WHERE b.bundle_id=:p1
 63"""
 64
 65
 66GET_CATEGORIES = """-- name: get_categories \\:many
 67SELECT category_id, category_name, category_coefficient
 68FROM category
 69"""
 70
 71
 72GET_CATEGORY = """-- name: get_category \\:one
 73SELECT category_id, category_name, category_coefficient
 74FROM category
 75WHERE category_id=:p1
 76LIMIT 1
 77"""
 78
 79
 80UPDATE_CATEGORY = """-- name: update_category \\:one
 81UPDATE category
 82SET category_name = :p2, category_coefficient = :p3
 83WHERE category_id = :p1
 84RETURNING category_id, category_name, category_coefficient
 85"""
 86
 87
 88class UpdateCategoryParams(pydantic.BaseModel):
 89    category_id: int
 90    category_name: str
 91    category_coefficient: float
 92
 93
 94class AsyncQuerier:
 95    def __init__(self, conn: sqlalchemy.ext.asyncio.AsyncConnection):
 96        self._conn = conn
 97
 98    async def add_bundles_category(self, arg: AddBundlesCategoryParams) -> Optional[models.BundleCategory]:
 99        row = (await self._conn.execute(sqlalchemy.text(ADD_BUNDLES_CATEGORY), {"p1": arg.bundle_id, "p2": arg.category_id})).first()
100        if row is None:
101            return None
102        return models.BundleCategory(
103            category_id=row[0],
104            bundle_id=row[1],
105        )
106
107    async def create_category(self, arg: CreateCategoryParams) -> Optional[models.Category]:
108        row = (await self._conn.execute(sqlalchemy.text(CREATE_CATEGORY), {"p1": arg.category_name, "p2": arg.category_coefficient})).first()
109        if row is None:
110            return None
111        return models.Category(
112            category_id=row[0],
113            category_name=row[1],
114            category_coefficient=row[2],
115        )
116
117    async def delete_bundle_category(self, arg: DeleteBundleCategoryParams) -> Optional[models.BundleCategory]:
118        row = (await self._conn.execute(sqlalchemy.text(DELETE_BUNDLE_CATEGORY), {"p1": arg.category_id, "p2": arg.bundle_id})).first()
119        if row is None:
120            return None
121        return models.BundleCategory(
122            category_id=row[0],
123            bundle_id=row[1],
124        )
125
126    async def delete_category(self, *, category_id: int) -> Optional[models.Category]:
127        row = (await self._conn.execute(sqlalchemy.text(DELETE_CATEGORY), {"p1": category_id})).first()
128        if row is None:
129            return None
130        return models.Category(
131            category_id=row[0],
132            category_name=row[1],
133            category_coefficient=row[2],
134        )
135
136    async def get_bundle_categories(self, *, bundle_id: int) -> AsyncIterator[int]:
137        result = await self._conn.stream(sqlalchemy.text(GET_BUNDLE_CATEGORIES), {"p1": bundle_id})
138        async for row in result:
139            yield row[0]
140
141    async def get_categories(self) -> AsyncIterator[models.Category]:
142        result = await self._conn.stream(sqlalchemy.text(GET_CATEGORIES))
143        async for row in result:
144            yield models.Category(
145                category_id=row[0],
146                category_name=row[1],
147                category_coefficient=row[2],
148            )
149
150    async def get_category(self, *, category_id: int) -> Optional[models.Category]:
151        row = (await self._conn.execute(sqlalchemy.text(GET_CATEGORY), {"p1": category_id})).first()
152        if row is None:
153            return None
154        return models.Category(
155            category_id=row[0],
156            category_name=row[1],
157            category_coefficient=row[2],
158        )
159
160    async def update_category(self, arg: UpdateCategoryParams) -> Optional[models.Category]:
161        row = (await self._conn.execute(sqlalchemy.text(UPDATE_CATEGORY), {"p1": arg.category_id, "p2": arg.category_name, "p3": arg.category_coefficient})).first()
162        if row is None:
163            return None
164        return models.Category(
165            category_id=row[0],
166            category_name=row[1],
167            category_coefficient=row[2],
168        )