sqlc-gen-python
内容としては、SQLAlchemyを使って実行する形式となります。型も定義されるので親切設計になっています。
/icons/hr.icon
入力と出力のサンプルを以下に記載します。
入力
code:schema.sql
CREATE TABLE authors (
id UUID PRIMARY KEY,
name text NOT NULL,
bio text
);
create table books
(
id UUID PRIMARY KEY,
title varchar(99) not null,
price BIGSERIAL not null,
author_id UUID not null,
created_at timestamp not null default now()
);
alter table books add foreign key (author_id) references authosr (id);
code:authors.sql
-- name: GetAuthor :one
SELECT * FROM authors
WHERE id = $1 LIMIT 1;
-- name: ListAuthors :many
SELECT * FROM authors
ORDER BY name;
-- name: CreateAuthor :one
INSERT INTO authors (
id, name, bio
) VALUES (
$1, $2, $3
)
RETURNING *;
-- name: UpdateAuthor :exec
UPDATE authors
set name = $2,
bio = $3
WHERE id = $1;
-- name: DeleteAuthor :exec
DELETE FROM authors
WHERE id = $1;
code:books.sql
-- name: ListBookOverPrice :many
SELECT
b.title
, a.name
, b.price
FROM
books b
LEFT JOIN
authors a
ON 1 = 1
AND b.author_id = a.id
WHERE
price > $1
ORDER BY
b.title;
出力
code:model.py
# Code generated by sqlc. DO NOT EDIT.
# versions:
# sqlc v1.25.0
import dataclasses
import datetime
from typing import Optional
import uuid
@dataclasses.dataclass()
class Author:
id: uuid.UUID
name: str
@dataclasses.dataclass()
class Book:
id: uuid.UUID
title: str
price: int
author_id: uuid.UUID
created_at: datetime.datetime
code:authors.py
# Code generated by sqlc. DO NOT EDIT.
# versions:
# sqlc v1.25.0
# source: authors.sql
from typing import Iterator, Optional
import uuid
import sqlalchemy
from sql.code import models
CREATE_AUTHOR = """-- name: create_author \\:one
INSERT INTO authors (
id, name, bio
) VALUES (
:p1, :p2, :p3
)
RETURNING id, name, bio
"""
DELETE_AUTHOR = """-- name: delete_author \\:exec
DELETE FROM authors
WHERE id = :p1
"""
GET_AUTHOR = """-- name: get_author \\:one
SELECT id, name, bio FROM authors
WHERE id = :p1 LIMIT 1
"""
LIST_AUTHORS = """-- name: list_authors \\:many
SELECT id, name, bio FROM authors
ORDER BY name
"""
UPDATE_AUTHOR = """-- name: update_author \\:exec
UPDATE authors
set name = :p2,
bio = :p3
WHERE id = :p1
"""
class Querier:
def __init__(self, conn: sqlalchemy.engine.Connection):
self._conn = conn
def create_author(self, *, id: uuid.UUID, name: str, bio: Optionalstr) -> Optionalmodels.Author: row = self._conn.execute(sqlalchemy.text(CREATE_AUTHOR), {"p1": id, "p2": name, "p3": bio}).first()
if row is None:
return None
return models.Author(
)
def delete_author(self, *, id: uuid.UUID) -> None:
self._conn.execute(sqlalchemy.text(DELETE_AUTHOR), {"p1": id})
row = self._conn.execute(sqlalchemy.text(GET_AUTHOR), {"p1": id}).first()
if row is None:
return None
return models.Author(
)
result = self._conn.execute(sqlalchemy.text(LIST_AUTHORS))
for row in result:
yield models.Author(
)
def update_author(self, *, id: uuid.UUID, name: str, bio: Optionalstr) -> None: self._conn.execute(sqlalchemy.text(UPDATE_AUTHOR), {"p1": id, "p2": name, "p3": bio})
code:books.py
# Code generated by sqlc. DO NOT EDIT.
# versions:
# sqlc v1.25.0
# source: books.sql
import dataclasses
from typing import Iterator, Optional
import sqlalchemy
from sql.code import models
LIST_BOOK_OVER_PRICE = """-- name: list_book_over_price \\:many
SELECT
b.title
, a.name
, b.price
FROM
books b
LEFT JOIN
authors a
ON 1 = 1
AND b.author_id = a.id
WHERE
price > :p1
ORDER BY
b.title
"""
@dataclasses.dataclass()
class ListBookOverPriceRow:
title: str
price: int
class Querier:
def __init__(self, conn: sqlalchemy.engine.Connection):
self._conn = conn
result = self._conn.execute(sqlalchemy.text(LIST_BOOK_OVER_PRICE), {"p1": price})
for row in result:
yield ListBookOverPriceRow(
)
/icons/hr.icon
code:使い方.py
import uuid
import sqlalchemy
from sqlalchemy import create_engine
from sql.code import authors, books
engine = create_engine("postgresql://root:root@localhost:5432/root")
# 検索
with engine.connect() as connection:
querier = authors.Querier(conn=connection)
id = uuid.UUID("c4b74fcb-ae5c-4270-84df-c4cb8f24d37c")
author = querier.get_author(id=id)
# 作成
with engine.connect() as connection:
querier = authors.Querier(conn=connection)
id = uuid.uuid4()
name = "著者名"
bio = "bio"
connection.begin()
created_author = querier.create_author(id=id, name=name, bio=bio)
print(created_author)
connection.commit()