/
Lepecin
/
newtodoapp
Обзор
Документация
Войти
/
Lepecin
/
newtodoapp
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
src/app/database.py
65 строк
2 KB
Lepecin
Init: Create project
06 фев 2026, 11:41
06 фев 2026, 11:41
d2c3fac
Код
Авторство
О чём код?
from psycopg import AsyncConnection from pydantic import BaseModel from datetime import datetime from .config import DATABASE_CONNECTION_INFO class InputMessage(BaseModel): message: str class Message(InputMessage): id: int time_created: datetime async def create_table(): async with await AsyncConnection.connect( conninfo=DATABASE_CONNECTION_INFO ) as connection: await connection.execute(""" create table if not exists messages ( id serial primary key, message text, time_created timestamptz default now() ) """) async def create_message(message: InputMessage): async with await AsyncConnection.connect( conninfo=DATABASE_CONNECTION_INFO ) as connection: await connection.execute( """ insert into messages(message) values (%s) """, params=(message.message,), ) async def get_messages(): async with await AsyncConnection.connect( conninfo=DATABASE_CONNECTION_INFO ) as connection: cursor = await connection.execute( "select id, message, time_created from messages" ) async for row in cursor: yield Message(id=row[0], message=row[1], time_created=row[2]) async def remove_message(id: int): async with await AsyncConnection.connect( conninfo=DATABASE_CONNECTION_INFO ) as connection: await connection.execute( """ delete from messages where id = %s """, params=(id,), )