Привет! Сегодня я хочу поделиться с вами своим опытом создания приложения на FastAPI с использованием базы данных Postgres, а также реализации CRUD-операций, аутентификации и авторизации. Я также покажу, как создать роли ″Модератор″ и ″Пользователь″ и использовать асинхронность. Поехали!1. Установка необходимых инструментов
Первым шагом я установил Python, FastAPI и Postgres. Для установки FastAPI и Postgres можно использовать pip⁚
pip install fastapi
pip install databases[postgresql]
2. Настройка базы данных
Затем я создал базу данных Postgres и таблицу ″Users″. В файле `app.py` я добавил следующий код⁚
python
import databases
import sqlalchemy
from sqlalchemy import Column, Integer, String
DATABASE_URL ″postgresql://username:password@localhost/database_name»
database databases.Database(DATABASE_URL)
metadata sqlalchemy.MetaData
users sqlalchemy.Table(
″users″,
metadata,
Column(″id″, Integer, primary_keyTrue),
Column(″username″, String),
Column(″password″, String),
Column(″role″, String),
)
engine sqlalchemy.create_engine(DATABASE_URL)
metadata.create_all(engine)
3. Создание CRUD-операций
Далее я создал файл `crud.py` для реализации операций CRUD. В нем я добавил код для создания, чтения, обновления и удаления пользователей⁚
python
from typing import List, Optional
from sqlalchemy import select
from app import database, users
async def create_user(username⁚ str, password⁚ str, role⁚ str)⁚
query users.insert.values(usernameusername, passwordpassword, rolerole)
await database.execute(query)
async def get_user(user_id⁚ int)⁚
query users.select.where(users.c.id user_id)
return await database.fetch_one(query)
async def get_users⁚
query users.select
return await database.fetch_all(query)
async def update_user(user_id⁚ int, username⁚ Optional[str] None,
password⁚ Optional[str] None, role⁚ Optional[str] None)⁚
query users.update.where(users.c.id user_id)
if username⁚
query query.values(usernameusername)
if password⁚
query query.values(passwordpassword)
if role⁚
query query.values(rolerole)
await database.execute(query)
async def delete_user(user_id⁚ int)⁚
query users.delete.where(users.c.id user_id)
await database.execute(query)
4. Аутентификация и авторизация
Для реализации аутентификации и авторизации я использовал библиотеку `Passlib`. Я создал файл `security.py` и добавил в него следующий код⁚
python
from passlib.context import CryptContext
pwd_context CryptContext(schemes[″bcrypt″], deprecated″auto″)
def verify_password(plain_password, hashed_password)⁚
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password)⁚
return pwd_context.hash(password)
5. Создание ролей ″Модератор″ и ″Пользователь″
Для создания ролей я добавил соответствующий столбец в таблицу ″Users″. В файле `crud.py` я также добавил функцию для получения пользователей с определенной ролью⁚
python
async def get_users_by_role(role⁚ str)⁚
query users.select.where(users.c.role role)
return await database.fetch_all(query)
6. Использование асинхронности
FastAPI поддерживает асинхронность из коробки. Я изменил функции CRUD-операций, добавив `async` и `await`⁚
python
async def create_user(username⁚ str, password⁚ str, role⁚ str)⁚
...async def get_user(user_id⁚ int)⁚
...async def get_users⁚
.;.async def update_user(user_id⁚ int, username⁚ Optional[str] None,
password⁚ Optional[str] None, role⁚ Optional[str] None)⁚
...async def delete_user(user_id⁚ int)⁚
...async def get_users_by_role(role⁚ str)⁚
...7. Пример кода приложения
В файле `main.py` я добавил следующий код для создания API с использованием FastAPI⁚
python
from fastapi import FastAPI, Depends
from app import database, engine, metadata
from crud import create_user, get_user, get_users, update_user, delete_user, get_users_by_role
from models import UserCreate, UserUpdate
from security import get_password_hash, verify_password
app FastAPI
metadata.create_all(bindengine)
@app.on_event(″startup″)
async def startup⁚
await database.connect
@app.on_event(″shutdown″)
async def shutdown⁚
await database.disconnect
@app.post(″/users»)
async def create_new_user(user⁚ UserCreate)⁚
hashed_password get_password_hash(user.password)
await create_user(user.username, hashed_password, user.role)
return {″message″⁚ ″User created″}
@app.get(″/users/{user_id}»)
async def get_single_user(user_id⁚ int)⁚
return await get_user(user_id)
@app.get(″/users»)
async def get_all_users⁚
return await get_users
@app.put(″/users/{user_id}»)
async def update_existing_user(user_id⁚ int, user⁚ UserUpdate)⁚
await update_user(user_id, user.username, user.password, user.role)
return {″message″⁚ ″User updated″}
@app.delete(″/users/{user_id}»)
async def delete_existing_user(user_id⁚ int)⁚
await delete_user(user_id)
return {″message″⁚ ″User deleted″}
@app.get(″/users/by_role/{role}»)
async def get_users_by_role_handler(role⁚ str)⁚
return await get_users_by_role(role)
Это лишь пример кода, и в реальном проекте возможно потребуются дополнительные проверки и обработка ошибок. Однако, эти шаги помогут вам начать создавать приложение на FastAPI с использованием базы данных Postgres, реализовать CRUD-операции, аутентификацию и авторизацию, а также использовать асинхронность.
Я надеюсь, что эта статья была полезной! Удачи в разработке!