Привет, я Максим, и я хочу поделиться с тобой своим опытом подключения к базе данных Postgres с использованием фреймворка FastAPI и асинхронного движка.Во-первых, чтобы начать работу, нам понадобится установить некоторые зависимости. Убедитесь, что вы установили библиотеки `fastapi`, `uvicorn`, `sqlalchemy`, `asyncpg` и `alembic`. Если они установлены, мы можем перейти к написанию кода.python
# main.py
from fastapi import FastAPI
from sqlalchemy import create_engine, Column, Integer, String, DateTime, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime
from asyncpg.exceptions import DuplicateTableError
# Подключение к базе данных
engine create_engine(″postgresql asyncpg://username:password@localhost/db_name»)
SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine)
Base declarative_base
# Создание модели
class Task(Base)⁚
__tablename__ ‘tasks’
id Column(Integer, primary_keyTrue, indexTrue)
title Column(String, nullableFalse, uniqueTrue)
description Column(String, nullableFalse, uniqueTrue)
created_date Column(DateTime, defaultdatetime.utcnow)
is_completed Column(Boolean, defaultFalse)
# Создание таблицы, если она не существует
try⁚
Base.metadata.create_all(bindengine)
except DuplicateTableError⁚
pass
# Создание FastAPI приложения
app FastAPI
# Роут для создания задачи
@app.post(″/tasks/»)
async def create_task(title⁚ str, description⁚ str)⁚
db SessionLocal
task Task(titletitle, descriptiondescription)
db.add(task)
db.commit
db.refresh(task)
return task
Давайте разберем этот код. Сначала мы подключаем необходимые модули и создаем движок для подключения к базе данных Postgres. Здесь мы указываем URL соединения, заменив `username`, `password` и `db_name` на ваши реальные данные. Затем мы создаем класс модели `Task`, который наследуется от `Base` из SQLAlchemy. В нем мы определяем поля таблицы и их типы данных.
Чтобы создать таблицу, мы используем метод `create_all` из `metadata` объекта, привязанного к движку. Однако, прежде чем вызывать этот метод, мы оборачиваем его в блок try-except, чтобы избежать создания дубликата таблицы. Если таблица уже существует, мы просто проигнорируем ошибку и продолжим выполнение программы.
Затем мы создаем экземпляр FastAPI и определяем маршрут для создания задачи. В этом маршруте мы создаем экземпляр сессии, добавляем новую задачу в сессию, коммитим изменения и обновляем объект задачи. Наконец, мы возвращаем созданную задачу.
Теперь, когда код готов, мы можем запустить сервер FastAPI с помощью команды `uvicorn main⁚app —reload`. После запуска сервера мы можем выполнить POST-запрос, отправив данные на `/tasks/`, и новая задача будет добавлена в базу данных.
Это подробный код для подключения к базе данных Postgres на асинхронном движке с использованием FastAPI и создания таблицы Задачи на основе модели Задачи. Надеюсь, эта статья и код будут полезными для тебя в твоих проектах!