-
Notifications
You must be signed in to change notification settings - Fork 4
Expand file tree
/
Copy pathqueries.py
More file actions
150 lines (114 loc) · 3.59 KB
/
queries.py
File metadata and controls
150 lines (114 loc) · 3.59 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
# Code generated by sqlc. DO NOT EDIT.
# versions:
# sqlc v1.28.0
# sqlc-gen-better-python v0.0.1
from __future__ import annotations
__all__: typing.Sequence[str] = (
"CreateAuthorParams",
"GetAuthorRow",
"UpdateAuthorParams",
"UpdateAuthorTParams",
"UpsertAuthorNameParams",
"create_author",
"delete_author",
"get_author",
"list_authors",
"update_author",
"update_author_t",
"upsert_author_name",
)
import dataclasses
import typing
import sqlite3
from test import models
@dataclasses.dataclass()
class CreateAuthorParams:
name: str
bio: typing.Optional[str]
@dataclasses.dataclass()
class GetAuthorRow:
id: int
name: str
@dataclasses.dataclass()
class UpdateAuthorParams:
name: str
bio: typing.Optional[str]
id: int
@dataclasses.dataclass()
class UpdateAuthorTParams:
name: typing.Optional[str]
bio: typing.Optional[str]
id: int
@dataclasses.dataclass()
class UpsertAuthorNameParams:
set_name: str
name: str
CREATE_AUTHOR: typing.Final[str] = """-- name: CreateAuthor :one
INSERT INTO authors (
name, bio
) VALUES (
?, ?
)
RETURNING id, name, bio
"""
DELETE_AUTHOR: typing.Final[str] = """-- name: DeleteAuthor :exec
DELETE FROM authors
WHERE id = ?
"""
GET_AUTHOR: typing.Final[str] = """-- name: GetAuthor :one
SELECT id, name FROM authors
WHERE id = ? LIMIT 1
"""
LIST_AUTHORS: typing.Final[str] = """-- name: ListAuthors :many
SELECT id, name, bio FROM authors
WHERE id IN (/*SLICE:ids*/?)
ORDER BY name
"""
UPDATE_AUTHOR: typing.Final[str] = """-- name: UpdateAuthor :exec
UPDATE authors
set name = ?,
bio = ?
WHERE id = ?
"""
UPDATE_AUTHOR_T: typing.Final[str] = """-- name: UpdateAuthorT :one
UPDATE authors
SET
name = coalesce(?1, name),
bio = coalesce(?2, bio)
WHERE id = ?3
RETURNING id, name, bio
"""
UPSERT_AUTHOR_NAME: typing.Final[str] = """-- name: UpsertAuthorName :one
UPDATE authors
SET name = CASE WHEN ?2 THEN ? ELSE name END
RETURNING id, name, bio
"""
def create_author(conn: sqlite3.Connection, arg: CreateAuthorParams) -> typing.Optional[models.Author]:
row = conn.execute(CREATE_AUTHOR,(arg.name, arg.bio, )).fetchone()
if row is None:
return None
return models.Author(id=row[0], name=row[1], bio=row[2])
def delete_author(conn: sqlite3.Connection, id: int) -> None:
conn.execute(DELETE_AUTHOR,(id, ))
def get_author(conn: sqlite3.Connection, id: int) -> typing.Optional[GetAuthorRow]:
row = conn.execute(GET_AUTHOR,(id, )).fetchone()
if row is None:
return None
return GetAuthorRow(id=row[0], name=row[1])
def list_authors(conn: sqlite3.Connection, ids: typing.Sequence[int]) -> typing.List[models.Author]:
rows: typing.List[models.Author] = []
for row in conn.execute(LIST_AUTHORS,(ids, )).fetchall():
rows.append(models.Author(id=row[0], name=row[1], bio=row[2]))
return rows
def update_author(conn: sqlite3.Connection, arg: UpdateAuthorParams) -> None:
conn.execute(UPDATE_AUTHOR,(arg.name, arg.bio, arg.id, ))
def update_author_t(conn: sqlite3.Connection, arg: UpdateAuthorTParams) -> typing.Optional[models.Author]:
row = conn.execute(UPDATE_AUTHOR_T,(arg.name, arg.bio, arg.id, )).fetchone()
if row is None:
return None
return models.Author(id=row[0], name=row[1], bio=row[2])
def upsert_author_name(conn: sqlite3.Connection, arg: UpsertAuthorNameParams) -> typing.Optional[models.Author]:
row = conn.execute(UPSERT_AUTHOR_NAME,(arg.set_name, arg.name, )).fetchone()
if row is None:
return None
return models.Author(id=row[0], name=row[1], bio=row[2])