query.py

# Code generated by sqlc. DO NOT EDIT.
# versions:
#   sqlc v1.30.0
# source: query.sql
import dataclasses
from typing import Any, List, Optional

import sqlalchemy
import sqlalchemy.ext.asyncio

from copyfrom import models


CREATE_AUTHOR = """-- name: create_author \\:one
INSERT INTO authors (name, bio) VALUES (:p1, :p2) RETURNING id, name, bio
"""


CREATE_AUTHORS = """-- name: create_authors \\:copyfrom
INSERT INTO authors (name, bio) VALUES (:p1, :p2)
"""


CREATE_AUTHORS_NAMED = """-- name: create_authors_named \\:copyfrom
INSERT INTO authors (name, bio) VALUES (:p1, :p2)
"""


CREATE_USER = """-- name: create_user \\:one
INSERT INTO users (email, name) VALUES (:p1, :p2) RETURNING id, email, name, bio, age, active, created_at
"""


CREATE_USERS_BATCH = """-- name: create_users_batch \\:copyfrom
INSERT INTO users (email, name) VALUES (:p1, :p2)
"""


CREATE_USERS_SHUFFLED = """-- name: create_users_shuffled \\:copyfrom
INSERT INTO users (email, name, bio, age, active) VALUES (:p2, :p1, :p3, :p4, :p5)
"""


@dataclasses.dataclass()
class CreateUsersShuffledParams:
    name: str
    email: str
    bio: Optional[str]
    age: Optional[int]
    active: Optional[bool]


CREATE_USERS_WITH_DETAILS = """-- name: create_users_with_details \\:copyfrom
INSERT INTO users (email, name, bio, age, active) VALUES (:p1, :p2, :p3, :p4, :p5)
"""


@dataclasses.dataclass()
class CreateUsersWithDetailsParams:
    email: str
    name: str
    bio: Optional[str]
    age: Optional[int]
    active: Optional[bool]


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

    def create_author(self, *, name: str, bio: str) -> Optional[models.Author]:
        row = self._conn.execute(sqlalchemy.text(CREATE_AUTHOR), {"p1": name, "p2": bio}).first()
        if row is None:
            return None
        return models.Author(
            id=row[0],
            name=row[1],
            bio=row[2],
        )

    def create_authors(self, arg_list: List[Any]) -> int:
        result = self._conn.execute(sqlalchemy.text(CREATE_AUTHORS), arg_list)
        return result.rowcount

    def create_authors_named(self, arg_list: List[Any]) -> int:
        result = self._conn.execute(sqlalchemy.text(CREATE_AUTHORS_NAMED), arg_list)
        return result.rowcount

    def create_user(self, *, email: str, name: str) -> Optional[models.User]:
        row = self._conn.execute(sqlalchemy.text(CREATE_USER), {"p1": email, "p2": name}).first()
        if row is None:
            return None
        return models.User(
            id=row[0],
            email=row[1],
            name=row[2],
            bio=row[3],
            age=row[4],
            active=row[5],
            created_at=row[6],
        )

    def create_users_batch(self, arg_list: List[Any]) -> int:
        result = self._conn.execute(sqlalchemy.text(CREATE_USERS_BATCH), arg_list)
        return result.rowcount

    def create_users_shuffled(self, arg_list: List[CreateUsersShuffledParams]) -> int:
        data = list()
        for item in arg_list:
            data.append({
                "p1": item.name,
                "p2": item.email,
                "p3": item.bio,
                "p4": item.age,
                "p5": item.active,
            })
        result = self._conn.execute(sqlalchemy.text(CREATE_USERS_SHUFFLED), data)
        return result.rowcount

    def create_users_with_details(self, arg_list: List[CreateUsersWithDetailsParams]) -> int:
        data = list()
        for item in arg_list:
            data.append({
                "p1": item.email,
                "p2": item.name,
                "p3": item.bio,
                "p4": item.age,
                "p5": item.active,
            })
        result = self._conn.execute(sqlalchemy.text(CREATE_USERS_WITH_DETAILS), data)
        return result.rowcount


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

    async def create_author(self, *, name: str, bio: str) -> Optional[models.Author]:
        row = (await self._conn.execute(sqlalchemy.text(CREATE_AUTHOR), {"p1": name, "p2": bio})).first()
        if row is None:
            return None
        return models.Author(
            id=row[0],
            name=row[1],
            bio=row[2],
        )

    async def create_authors(self, arg_list: List[Any]) -> int:
        result = await self._conn.execute(sqlalchemy.text(CREATE_AUTHORS), arg_list)
        return result.rowcount

    async def create_authors_named(self, arg_list: List[Any]) -> int:
        result = await self._conn.execute(sqlalchemy.text(CREATE_AUTHORS_NAMED), arg_list)
        return result.rowcount

    async def create_user(self, *, email: str, name: str) -> Optional[models.User]:
        row = (await self._conn.execute(sqlalchemy.text(CREATE_USER), {"p1": email, "p2": name})).first()
        if row is None:
            return None
        return models.User(
            id=row[0],
            email=row[1],
            name=row[2],
            bio=row[3],
            age=row[4],
            active=row[5],
            created_at=row[6],
        )

    async def create_users_batch(self, arg_list: List[Any]) -> int:
        result = await self._conn.execute(sqlalchemy.text(CREATE_USERS_BATCH), arg_list)
        return result.rowcount

    async def create_users_shuffled(self, arg_list: List[CreateUsersShuffledParams]) -> int:
        data = list()
        for item in arg_list:
            data.append({
                "p1": item.name,
                "p2": item.email,
                "p3": item.bio,
                "p4": item.age,
                "p5": item.active,
            })
        result = await self._conn.execute(sqlalchemy.text(CREATE_USERS_SHUFFLED), data)
        return result.rowcount

    async def create_users_with_details(self, arg_list: List[CreateUsersWithDetailsParams]) -> int:
        data = list()
        for item in arg_list:
            data.append({
                "p1": item.email,
                "p2": item.name,
                "p3": item.bio,
                "p4": item.age,
                "p5": item.active,
            })
        result = await self._conn.execute(sqlalchemy.text(CREATE_USERS_WITH_DETAILS), data)
        return result.rowcount