internal.queries.analytics

  1# Code generated by sqlc. DO NOT EDIT.
  2# versions:
  3#   sqlc v1.30.0
  4# source: analytics.sql
  5import decimal
  6import pydantic
  7from typing import AsyncIterator, Optional
  8
  9import sqlalchemy
 10import sqlalchemy.ext.asyncio
 11
 12from internal.queries import models
 13
 14
 15CREATE_GRAPH = """-- name: create_graph \\:one
 16INSERT INTO analytics_graphs (seller_id, graph_type)
 17VALUES (:p1, :p2)
 18RETURNING graph_id, seller_id, graph_type, created_at
 19"""
 20
 21
 22class CreateGraphParams(pydantic.BaseModel):
 23    seller_id: int
 24    graph_type: int
 25
 26
 27CREATE_GRAPH_POINT = """-- name: create_graph_point \\:one
 28INSERT INTO analytics_points (series_id, sort_index, x, y)
 29VALUES (:p1, :p2, :p3, :p4)
 30RETURNING series_id, sort_index, x, y
 31"""
 32
 33
 34class CreateGraphPointParams(pydantic.BaseModel):
 35    series_id: int
 36    sort_index: int
 37    x: str
 38    y: decimal.Decimal
 39
 40
 41CREATE_GRAPH_SERIES = """-- name: create_graph_series \\:one
 42INSERT INTO analytics_series (graph_id, series_name, sort_index)
 43VALUES (:p1, :p2, :p3)
 44RETURNING series_id, graph_id, series_name, sort_index
 45"""
 46
 47
 48class CreateGraphSeriesParams(pydantic.BaseModel):
 49    graph_id: int
 50    series_name: str
 51    sort_index: int
 52
 53
 54DELETE_GRAPH_SERIES = """-- name: delete_graph_series \\:exec
 55DELETE FROM analytics_series
 56WHERE graph_id=:p1
 57"""
 58
 59
 60GET_GRAPH = """-- name: get_graph \\:one
 61SELECT graph_id, seller_id, graph_type, created_at
 62FROM analytics_graphs
 63WHERE seller_id=:p1 AND graph_type=:p2
 64LIMIT 1
 65"""
 66
 67
 68class GetGraphParams(pydantic.BaseModel):
 69    seller_id: int
 70    graph_type: int
 71
 72
 73GET_GRAPH_POINTS = """-- name: get_graph_points \\:many
 74SELECT series_id, sort_index, x, y
 75FROM analytics_points
 76WHERE series_id=:p1
 77"""
 78
 79
 80GET_GRAPH_SERIES = """-- name: get_graph_series \\:many
 81SELECT series_id, graph_id, series_name, sort_index
 82FROM analytics_series
 83WHERE graph_id=:p1
 84ORDER BY sort_index
 85"""
 86
 87
 88GET_GRAPH_TYPE = """-- name: get_graph_type \\:one
 89SELECT graph_type_id, chart_type, graph_summary, x_axis_label, y_axis_label
 90FROM analytics_graphs_types
 91WHERE graph_type_id=:p1
 92"""
 93
 94
 95GET_GRAPHS = """-- name: get_graphs \\:many
 96SELECT graph_id, seller_id, graph_type, created_at
 97FROM analytics_graphs
 98WHERE seller_id=:p1
 99"""
100
101
102GET_GRAPHS_TYPES = """-- name: get_graphs_types \\:many
103SELECT graph_type_id, chart_type, graph_summary, x_axis_label, y_axis_label
104FROM analytics_graphs_types
105"""
106
107
108class AsyncQuerier:
109    def __init__(self, conn: sqlalchemy.ext.asyncio.AsyncConnection):
110        self._conn = conn
111
112    async def create_graph(self, arg: CreateGraphParams) -> Optional[models.AnalyticsGraph]:
113        row = (await self._conn.execute(sqlalchemy.text(CREATE_GRAPH), {"p1": arg.seller_id, "p2": arg.graph_type})).first()
114        if row is None:
115            return None
116        return models.AnalyticsGraph(
117            graph_id=row[0],
118            seller_id=row[1],
119            graph_type=row[2],
120            created_at=row[3],
121        )
122
123    async def create_graph_point(self, arg: CreateGraphPointParams) -> Optional[models.AnalyticsPoint]:
124        row = (await self._conn.execute(sqlalchemy.text(CREATE_GRAPH_POINT), {
125            "p1": arg.series_id,
126            "p2": arg.sort_index,
127            "p3": arg.x,
128            "p4": arg.y,
129        })).first()
130        if row is None:
131            return None
132        return models.AnalyticsPoint(
133            series_id=row[0],
134            sort_index=row[1],
135            x=row[2],
136            y=row[3],
137        )
138
139    async def create_graph_series(self, arg: CreateGraphSeriesParams) -> Optional[models.AnalyticsSeries]:
140        row = (await self._conn.execute(sqlalchemy.text(CREATE_GRAPH_SERIES), {"p1": arg.graph_id, "p2": arg.series_name, "p3": arg.sort_index})).first()
141        if row is None:
142            return None
143        return models.AnalyticsSeries(
144            series_id=row[0],
145            graph_id=row[1],
146            series_name=row[2],
147            sort_index=row[3],
148        )
149
150    async def delete_graph_series(self, *, graph_id: int) -> None:
151        await self._conn.execute(sqlalchemy.text(DELETE_GRAPH_SERIES), {"p1": graph_id})
152
153    async def get_graph(self, arg: GetGraphParams) -> Optional[models.AnalyticsGraph]:
154        row = (await self._conn.execute(sqlalchemy.text(GET_GRAPH), {"p1": arg.seller_id, "p2": arg.graph_type})).first()
155        if row is None:
156            return None
157        return models.AnalyticsGraph(
158            graph_id=row[0],
159            seller_id=row[1],
160            graph_type=row[2],
161            created_at=row[3],
162        )
163
164    async def get_graph_points(self, *, series_id: int) -> AsyncIterator[models.AnalyticsPoint]:
165        result = await self._conn.stream(sqlalchemy.text(GET_GRAPH_POINTS), {"p1": series_id})
166        async for row in result:
167            yield models.AnalyticsPoint(
168                series_id=row[0],
169                sort_index=row[1],
170                x=row[2],
171                y=row[3],
172            )
173
174    async def get_graph_series(self, *, graph_id: int) -> AsyncIterator[models.AnalyticsSeries]:
175        result = await self._conn.stream(sqlalchemy.text(GET_GRAPH_SERIES), {"p1": graph_id})
176        async for row in result:
177            yield models.AnalyticsSeries(
178                series_id=row[0],
179                graph_id=row[1],
180                series_name=row[2],
181                sort_index=row[3],
182            )
183
184    async def get_graph_type(self, *, graph_type_id: int) -> Optional[models.AnalyticsGraphsType]:
185        row = (await self._conn.execute(sqlalchemy.text(GET_GRAPH_TYPE), {"p1": graph_type_id})).first()
186        if row is None:
187            return None
188        return models.AnalyticsGraphsType(
189            graph_type_id=row[0],
190            chart_type=row[1],
191            graph_summary=row[2],
192            x_axis_label=row[3],
193            y_axis_label=row[4],
194        )
195
196    async def get_graphs(self, *, seller_id: int) -> AsyncIterator[models.AnalyticsGraph]:
197        result = await self._conn.stream(sqlalchemy.text(GET_GRAPHS), {"p1": seller_id})
198        async for row in result:
199            yield models.AnalyticsGraph(
200                graph_id=row[0],
201                seller_id=row[1],
202                graph_type=row[2],
203                created_at=row[3],
204            )
205
206    async def get_graphs_types(self) -> AsyncIterator[models.AnalyticsGraphsType]:
207        result = await self._conn.stream(sqlalchemy.text(GET_GRAPHS_TYPES))
208        async for row in result:
209            yield models.AnalyticsGraphsType(
210                graph_type_id=row[0],
211                chart_type=row[1],
212                graph_summary=row[2],
213                x_axis_label=row[3],
214                y_axis_label=row[4],
215            )