63 lines
No EOL
1.6 KiB
Python
63 lines
No EOL
1.6 KiB
Python
from pydantic import BaseModel
|
|
from typing import AsyncIterator, ClassVar, List
|
|
|
|
class Person(BaseModel):
|
|
KEYS: ClassVar[List[str]] = ['id', 'name']
|
|
|
|
id: int
|
|
name: str
|
|
|
|
async def create(conn):
|
|
await conn.execute('''
|
|
CREATE TABLE IF NOT EXISTS Person (
|
|
id INTEGER PRIMARY KEY,
|
|
name TEXT UNIQUE
|
|
);''')
|
|
|
|
async def search_by_name(conn, name: str) -> AsyncIterator[Person]:
|
|
async with conn.execute('''
|
|
SELECT id, name
|
|
FROM Person
|
|
WHERE name LIKE ?
|
|
''', (f'%{name}%',)) as cursor:
|
|
async for row in cursor:
|
|
yield Person(id=row[0], name=row[1])
|
|
|
|
|
|
async def get_by_name(conn, name: str) -> Person:
|
|
cursor = await conn.execute('''
|
|
SELECT id, name
|
|
FROM Person
|
|
WHERE name = ?
|
|
''', (name,))
|
|
row = await cursor.fetchone()
|
|
if not row:
|
|
return None
|
|
return Person(id=row[0], name=row[1])
|
|
|
|
async def get_by_id(conn, id: int) -> Person:
|
|
cursor = await conn.execute('''
|
|
SELECT id, name
|
|
FROM Person
|
|
WHERE id = ?
|
|
''', (id,))
|
|
row = await cursor.fetchone()
|
|
if not row:
|
|
return None
|
|
return Person(id=row[0], name=row[1])
|
|
|
|
async def get_all(conn) -> AsyncIterator[Person]:
|
|
async with conn.execute('''
|
|
SELECT id, name
|
|
FROM Person
|
|
''') as cursor:
|
|
async for row in cursor:
|
|
yield Person(id=row[0], name=row[1])
|
|
|
|
async def insert_person(conn, person: Person) -> Person:
|
|
cursor = await conn.execute('''
|
|
INSERT INTO Person (name)
|
|
VALUES (?)
|
|
''', (person.name,))
|
|
person.id = cursor.lastrowid
|
|
return person |