internal.queries.allergens

  1# Code generated by sqlc. DO NOT EDIT.
  2# versions:
  3#   sqlc v1.30.0
  4# source: allergens.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_ALLERGEN = """-- name: add_bundles_allergen \\:one
 15INSERT INTO bundle_allergens (bundle_id, allergen_id)
 16VALUES (:p1,:p2)
 17RETURNING bundle_id, allergen_id
 18"""
 19
 20
 21class AddBundlesAllergenParams(pydantic.BaseModel):
 22    bundle_id: int
 23    allergen_id: int
 24
 25
 26CREATE_ALLERGEN = """-- name: create_allergen \\:one
 27INSERT INTO allergens (allergen_name)
 28VALUES (:p1)
 29RETURNING allergen_id, allergen_name
 30"""
 31
 32
 33DELETE_ALLERGEN = """-- name: delete_allergen \\:one
 34DELETE FROM allergens
 35WHERE allergen_id = :p1
 36RETURNING allergen_id, allergen_name
 37"""
 38
 39
 40DELETE_BUNDLE_ALLERGEN = """-- name: delete_bundle_allergen \\:one
 41DELETE FROM bundle_allergens
 42WHERE allergen_id=:p1 AND bundle_id=:p2
 43RETURNING bundle_id, allergen_id
 44"""
 45
 46
 47class DeleteBundleAllergenParams(pydantic.BaseModel):
 48    allergen_id: int
 49    bundle_id: int
 50
 51
 52GET_ALLERGENS = """-- name: get_allergens \\:many
 53SELECT allergen_id, allergen_name
 54FROM allergens
 55"""
 56
 57
 58GET_BUNDLE_ALLERGENS = """-- name: get_bundle_allergens \\:many
 59SELECT a.allergen_id
 60FROM allergens a
 61JOIN bundle_allergens ba ON ba.allergen_id = a.allergen_id
 62JOIN bundles b ON b.bundle_id = ba.bundle_id
 63WHERE b.bundle_id=:p1
 64"""
 65
 66
 67UPDATE_ALLERGEN = """-- name: update_allergen \\:one
 68UPDATE allergens
 69SET allergen_name = :p2
 70WHERE allergen_id = :p1
 71RETURNING allergen_id, allergen_name
 72"""
 73
 74
 75class UpdateAllergenParams(pydantic.BaseModel):
 76    allergen_id: int
 77    allergen_name: str
 78
 79
 80class AsyncQuerier:
 81    def __init__(self, conn: sqlalchemy.ext.asyncio.AsyncConnection):
 82        self._conn = conn
 83
 84    async def add_bundles_allergen(self, arg: AddBundlesAllergenParams) -> Optional[models.BundleAllergen]:
 85        row = (await self._conn.execute(sqlalchemy.text(ADD_BUNDLES_ALLERGEN), {"p1": arg.bundle_id, "p2": arg.allergen_id})).first()
 86        if row is None:
 87            return None
 88        return models.BundleAllergen(
 89            bundle_id=row[0],
 90            allergen_id=row[1],
 91        )
 92
 93    async def create_allergen(self, *, allergen_name: str) -> Optional[models.Allergen]:
 94        row = (await self._conn.execute(sqlalchemy.text(CREATE_ALLERGEN), {"p1": allergen_name})).first()
 95        if row is None:
 96            return None
 97        return models.Allergen(
 98            allergen_id=row[0],
 99            allergen_name=row[1],
100        )
101
102    async def delete_allergen(self, *, allergen_id: int) -> Optional[models.Allergen]:
103        row = (await self._conn.execute(sqlalchemy.text(DELETE_ALLERGEN), {"p1": allergen_id})).first()
104        if row is None:
105            return None
106        return models.Allergen(
107            allergen_id=row[0],
108            allergen_name=row[1],
109        )
110
111    async def delete_bundle_allergen(self, arg: DeleteBundleAllergenParams) -> Optional[models.BundleAllergen]:
112        row = (await self._conn.execute(sqlalchemy.text(DELETE_BUNDLE_ALLERGEN), {"p1": arg.allergen_id, "p2": arg.bundle_id})).first()
113        if row is None:
114            return None
115        return models.BundleAllergen(
116            bundle_id=row[0],
117            allergen_id=row[1],
118        )
119
120    async def get_allergens(self) -> AsyncIterator[models.Allergen]:
121        result = await self._conn.stream(sqlalchemy.text(GET_ALLERGENS))
122        async for row in result:
123            yield models.Allergen(
124                allergen_id=row[0],
125                allergen_name=row[1],
126            )
127
128    async def get_bundle_allergens(self, *, bundle_id: int) -> AsyncIterator[int]:
129        result = await self._conn.stream(sqlalchemy.text(GET_BUNDLE_ALLERGENS), {"p1": bundle_id})
130        async for row in result:
131            yield row[0]
132
133    async def update_allergen(self, arg: UpdateAllergenParams) -> Optional[models.Allergen]:
134        row = (await self._conn.execute(sqlalchemy.text(UPDATE_ALLERGEN), {"p1": arg.allergen_id, "p2": arg.allergen_name})).first()
135        if row is None:
136            return None
137        return models.Allergen(
138            allergen_id=row[0],
139            allergen_name=row[1],
140        )