query.py

# Code generated by sqlc. DO NOT EDIT.
# versions:
#   sqlc v1.27.0
# source: query.sql
from typing import AsyncIterator, Iterator

import sqlalchemy
import sqlalchemy.ext.asyncio

from db import models


LIST = """-- name: list \\:many
SELECT a, b, operation FROM operations
"""


class Querier:
    def __init__(self, conn: sqlalchemy.engine.Connection):
        self._conn = conn

    def list(self) -> Iterator[models.Operation]:
        result = self._conn.execute(sqlalchemy.text(LIST))
        for row in result:
            yield models.Operation(
                a=row[0],
                b=row[1],
                operation=row[2],
            )


class AsyncQuerier:
    def __init__(self, conn: sqlalchemy.ext.asyncio.AsyncConnection):
        self._conn = conn

    async def list(self) -> AsyncIterator[models.Operation]:
        result = await self._conn.stream(sqlalchemy.text(LIST))
        async for row in result:
            yield models.Operation(
                a=row[0],
                b=row[1],
                operation=row[2],
            )