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
|
import dataclasses
import datetime
import ipaddress as ipaddr
import uuid
import asyncpg
import quart as q
def get_db() -> asyncpg.Pool:
return q.current_app.pool # pyright: ignore[reportAttributeAccessIssue]
@dataclasses.dataclass
class BlogPost:
id: uuid.UUID
title: str
description: str | None
text: str
original: str
created_at: datetime.datetime
public: bool
@classmethod
def from_record(cls, record: asyncpg.Record):
return cls(
id=record["id"],
title=record["title"],
description=record.get("description"),
text=record["text"],
original=record["original"],
created_at=record["created_at"],
public=record["public"],
)
@dataclasses.dataclass
class BlogComment:
id: uuid.UUID
author: str
text: str
created_at: datetime.datetime
post: uuid.UUID | None = None
video: uuid.UUID | None = None
@classmethod
def from_record(cls, record: asyncpg.Record):
return cls(
id=record["id"],
author=record["author"],
text=record["text"],
created_at=record["created_at"],
post=record.get("post_id"),
video=record.get("video_id"),
)
@dataclasses.dataclass
class AdminLogin:
id: int
addr: ipaddr.IPv4Interface | ipaddr.IPv6Interface
created_at: datetime.datetime
@classmethod
def from_record(cls, record: asyncpg.Record):
return cls(
id=record["id"],
addr=record["ipaddr"],
created_at=record["created_at"],
)
async def get_post(post: str, *, public_only: bool = True):
query = "SELECT * FROM posts WHERE posts.id = $1"
if public_only:
query += " AND posts.public = true"
query += ";"
pool = get_db()
async with pool.acquire() as conn:
res = await conn.fetch(query, post)
if not res:
return None
return BlogPost.from_record(res[0])
async def get_posts(*, public_only: bool = True):
query = "SELECT * FROM posts"
if public_only:
query += " WHERE posts.public = true"
query += ";"
pool = get_db()
async with pool.acquire() as conn:
res = await conn.fetch(query)
return [BlogPost.from_record(r) for r in res]
async def get_post_comments(post: str):
pool = get_db()
async with pool.acquire() as conn:
res = await conn.fetch(
"SELECT * FROM comments WHERE comments.post_id = $1 ORDER BY comments.created_at",
post,
)
finals = [BlogComment.from_record(r) for r in res]
return finals
|
