Пишем веб сервис на Python с помощью FastAPI

Схема базы данных и миграции

import sqlalchemyfrom sqlalchemy.dialects.postgresql import UUIDmetadata = sqlalchemy.MetaData()users_table = sqlalchemy.Table(    "users",    metadata,    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),    sqlalchemy.Column("email", sqlalchemy.String(40), unique=True, index=True),    sqlalchemy.Column("name", sqlalchemy.String(100)),    sqlalchemy.Column("hashed_password", sqlalchemy.String()),    sqlalchemy.Column(        "is_active",        sqlalchemy.Boolean(),        server_default=sqlalchemy.sql.expression.true(),        nullable=False,    ),)tokens_table = sqlalchemy.Table(    "tokens",    metadata,    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),    sqlalchemy.Column(        "token",        UUID(as_uuid=False),        server_default=sqlalchemy.text("uuid_generate_v4()"),        unique=True,        nullable=False,        index=True,    ),    sqlalchemy.Column("expires", sqlalchemy.DateTime()),    sqlalchemy.Column("user_id", sqlalchemy.ForeignKey("users.id")),)

import sqlalchemyfrom .users import users_tablemetadata = sqlalchemy.MetaData()posts_table = sqlalchemy.Table(    "posts",    metadata,    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),    sqlalchemy.Column("user_id", sqlalchemy.ForeignKey(users_table.c.id)),    sqlalchemy.Column("created_at", sqlalchemy.DateTime()),    sqlalchemy.Column("title", sqlalchemy.String(100)),    sqlalchemy.Column("content", sqlalchemy.Text()),)

$ pip install alembic

$ alembic init migrations

• каталог versions, в котором будут хранится файлы миграций
• скрипт env.py, запускающийся при вызове alembic
• файл script.py.mako, содержащий шаблон для новых миграций.

sqlalchemy.url = postgresql://%(DB_USER)s:%(DB_PASS)s@%(DB_HOST)s:5432/%(DB_NAME)s

from os import environfrom alembic import contextfrom app.models import posts, users# Alembic Config объект предоставляет доступ# к переменным из файла alembic.iniconfig = context.configsection = config.config_ini_sectionconfig.set_section_option(section, "DB_USER", environ.get("DB_USER"))config.set_section_option(section, "DB_PASS", environ.get("DB_PASS"))config.set_section_option(section, "DB_NAME", environ.get("DB_NAME"))config.set_section_option(section, "DB_HOST", environ.get("DB_HOST"))fileConfig(config.config_file_name)target_metadata = [users.metadata, posts.metadata]

$ alembic revision --autogenerate -m "Added required tables"$ alembic upgrade head

Запускаем приложение и подключаем БД

from fastapi import FastAPIapp = FastAPI()@app.get("/")def read_root():    return {"Hello": "World"}

$ uvicorn main:app --reload

{"Hello": "World"}

from os import environimport databases# берем параметры БД из переменных окруженияDB_USER = environ.get("DB_USER", "user")DB_PASSWORD = environ.get("DB_PASSWORD", "password")DB_HOST = environ.get("DB_HOST", "localhost")DB_NAME = "async-blogs"SQLALCHEMY_DATABASE_URL = (    f"postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:5432/{DB_NAME}")# создаем объект database, который будет использоваться для выполнения запросовdatabase = databases.Database(SQLALCHEMY_DATABASE_URL)app = FastAPI()@app.on_event("startup")async def startup():    # когда приложение запускается устанавливаем соединение с БД    await database.connect()@app.on_event("shutdown")async def shutdown():    # когда приложение останавливается разрываем соединение с БД    await database.disconnect()@app.get("/")def read_root():    # изменим роут таким образом, чтобы он брал данные из БД    query = (        select(            [                posts_table.c.id,                posts_table.c.created_at,                posts_table.c.title,                posts_table.c.content,                posts_table.c.user_id,                users_table.c.name.label("user_name"),            ]        )        .select_from(posts_table.join(users_table))        .order_by(desc(posts_table.c.created_at))    )    return await database.fetch_all(query)

Валидация запроса и ответа

pip install pydantic

from pydantic import BaseModel, EmailStrclass UserCreate(BaseModel):    """ Проверяет sign-up запрос """    email: EmailStr    name: str    password: str

pip install email-validator

from typing import Optionalfrom pydantic import UUID4, BaseModel, EmailStr, Field, validatorclass UserCreate(BaseModel):    """ Проверяет sign-up запрос """    email: EmailStr    name: str    password: strclass UserBase(BaseModel):    """ Формирует тело ответа с деталями пользователя """    id: int    email: EmailStr    name: strclass TokenBase(BaseModel):    token: UUID4 = Field(..., alias="access_token")    expires: datetime    token_type: Optional[str] = "bearer"    class Config:        allow_population_by_field_name = True    @validator("token")    def hexlify_token(cls, value):        """ Конвертирует UUID в hex строку """        return value.hexclass User(UserBase):    """ Формирует тело ответа с деталями пользователя и токеном """    token: TokenBase = {}

import hashlibimport randomimport stringfrom datetime import datetime, timedeltafrom sqlalchemy import and_from app.models.database import databasefrom app.models.users import tokens_table, users_tablefrom app.schemas import users as user_schemadef get_random_string(length=12):    """ Генерирует случайную строку, использующуюся как соль """    return "".join(random.choice(string.ascii_letters) for _ in range(length))def hash_password(password: str, salt: str = None):    """ Хеширует пароль с солью """    if salt is None:        salt = get_random_string()    enc = hashlib.pbkdf2_hmac("sha256", password.encode(), salt.encode(), 100_000)    return enc.hex()def validate_password(password: str, hashed_password: str):    """ Проверяет, что хеш пароля совпадает с хешем из БД """    salt, hashed = hashed_password.split("$")    return hash_password(password, salt) == hashedasync def get_user_by_email(email: str):    """ Возвращает информацию о пользователе """    query = users_table.select().where(users_table.c.email == email)    return await database.fetch_one(query)async def get_user_by_token(token: str):    """ Возвращает информацию о владельце указанного токена """    query = tokens_table.join(users_table).select().where(        and_(            tokens_table.c.token == token,            tokens_table.c.expires > datetime.now()        )    )    return await database.fetch_one(query)async def create_user_token(user_id: int):    """ Создает токен для пользователя с указанным user_id """    query = (        tokens_table.insert()        .values(expires=datetime.now() + timedelta(weeks=2), user_id=user_id)        .returning(tokens_table.c.token, tokens_table.c.expires)    )    return await database.fetch_one(query)async def create_user(user: user_schema.UserCreate):    """ Создает нового пользователя в БД """    salt = get_random_string()    hashed_password = hash_password(user.password, salt)    query = users_table.insert().values(        email=user.email, name=user.name, hashed_password=f"{salt}${hashed_password}"    )    user_id = await database.execute(query)    token = await create_user_token(user_id)    token_dict = {"token": token["token"], "expires": token["expires"]}    return {**user.dict(), "id": user_id, "is_active": True, "token": token_dict}

from fastapi import APIRouterfrom app.schemas import usersfrom app.utils import users as users_utilsrouter = APIRouter()@router.post("/sign-up", response_model=users.User)async def create_user(user: users.UserCreate):    db_user = await users_utils.get_user_by_email(email=user.email)    if db_user:        raise HTTPException(status_code=400, detail="Email already registered")    return await users_utils.create_user(user=user)

from app.routers import usersapp.include_router(users.router)

Аутентификация и контроль доступа

from fastapi import Dependsfrom fastapi.security import OAuth2PasswordRequestForm@router.post("/auth", response_model=users.TokenBase)async def auth(form_data: OAuth2PasswordRequestForm = Depends()):    user = await users_utils.get_user_by_email(email=form_data.username)    if not user:        raise HTTPException(status_code=400, detail="Incorrect email or password")    if not users_utils.validate_password(        password=form_data.password, hashed_password=user["hashed_password"]    ):        raise HTTPException(status_code=400, detail="Incorrect email or password")    return await users_utils.create_user_token(user_id=user["id"])

from app.utils import users as users_utilsfrom fastapi import Depends, HTTPException, statusfrom fastapi.security import OAuth2PasswordBeareroauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth")async def get_current_user(token: str = Depends(oauth2_scheme)):    user = await users_utils.get_user_by_token(token)    if not user:        raise HTTPException(            status_code=status.HTTP_401_UNAUTHORIZED,            detail="Invalid authentication credentials",            headers={"WWW-Authenticate": "Bearer"},        )    if not user["is_active"]:        raise HTTPException(            status_code=status.HTTP_400_BAD_REQUEST, detail="Inactive user"        )    return user

from app.utils.dependencies import get_current_user@router.get("/users/me", response_model=users.UserBase)async def read_users_me(current_user: users.User = Depends(get_current_user)):    return current_user

from datetime import datetimefrom app.models.database import databasefrom app.models.posts import posts_tablefrom app.models.users import users_tablefrom app.schemas import posts as post_schemafrom sqlalchemy import desc, func, selectasync def create_post(post: post_schema.PostModel, user):    query = (        posts_table.insert()        .values(            title=post.title,            content=post.content,            created_at=datetime.now(),            user_id=user["id"],        )        .returning(            posts_table.c.id,            posts_table.c.title,            posts_table.c.content,            posts_table.c.created_at,        )    )    post = await database.fetch_one(query)    # Convert to dict and add user_name key to it    post = dict(zip(post, post.values()))    post["user_name"] = user["name"]    return postasync def get_post(post_id: int):    query = (        select(            [                posts_table.c.id,                posts_table.c.created_at,                posts_table.c.title,                posts_table.c.content,                posts_table.c.user_id,                users_table.c.name.label("user_name"),            ]        )        .select_from(posts_table.join(users_table))        .where(posts_table.c.id == post_id)    )    return await database.fetch_one(query)async def get_posts(page: int):    max_per_page = 10    offset1 = (page - 1) * max_per_page    query = (        select(            [                posts_table.c.id,                posts_table.c.created_at,                posts_table.c.title,                posts_table.c.content,                posts_table.c.user_id,                users_table.c.name.label("user_name"),            ]        )        .select_from(posts_table.join(users_table))        .order_by(desc(posts_table.c.created_at))        .limit(max_per_page)        .offset(offset1)    )    return await database.fetch_all(query)async def get_posts_count():    query = select([func.count()]).select_from(posts_table)    return await database.fetch_val(query)async def update_post(post_id: int, post: post_schema.PostModel):    query = (        posts_table.update()        .where(posts_table.c.id == post_id)        .values(title=post.title, content=post.content)    )    return await database.execute(query)

from app.schemas.posts import PostDetailsModel, PostModelfrom app.schemas.users import Userfrom app.utils import posts as post_utilsfrom app.utils.dependencies import get_current_userfrom fastapi import APIRouter, Depends, HTTPException, statusrouter = APIRouter()@router.post("/posts", response_model=PostDetailsModel, status_code=201)async def create_post(post: PostModel, current_user: User = Depends(get_current_user)):    post = await post_utils.create_post(post, current_user)    return post@router.get("/posts")async def get_posts(page: int = 1):    total_cout = await post_utils.get_posts_count()    posts = await post_utils.get_posts(page)    return {"total_count": total_cout, "results": posts}@router.get("/posts/{post_id}", response_model=PostDetailsModel)async def get_post(post_id: int):    return await post_utils.get_post(post_id)@router.put("/posts/{post_id}", response_model=PostDetailsModel)async def update_post(    post_id: int, post_data: PostModel, current_user=Depends(get_current_user)):    post = await post_utils.get_post(post_id)    if post["user_id"] != current_user["id"]:        raise HTTPException(            status_code=status.HTTP_403_FORBIDDEN,            detail="You don't have access to modify this post",        )    await post_utils.update_post(post_id=post_id, post=post_data)    return await post_utils.get_post(post_id)

from app.routers import postsapp.include_router(posts.router)

Тестирование

$ pip install pytest

from app.main import appfrom fastapi.testclient import TestClientclient = TestClient(app)def test_health_check():    response = client.get("/")    assert response.status_code == 200    assert response.json() == {"Hello": "World"}

from os import environimport databasesDB_USER = environ.get("DB_USER", "user")DB_PASSWORD = environ.get("DB_PASSWORD", "password")DB_HOST = environ.get("DB_HOST", "localhost")TESTING = environ.get("TESTING")if TESTING:    # Используем отдельную базу данных для тестов    DB_NAME = "async-blogs-temp-for-test"    TEST_SQLALCHEMY_DATABASE_URL = (        f"postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:5432/{DB_NAME}"    )    database = databases.Database(TEST_SQLALCHEMY_DATABASE_URL)else:    DB_NAME = "async-blogs"    SQLALCHEMY_DATABASE_URL = (        f"postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:5432/{DB_NAME}"    )    database = databases.Database(SQLALCHEMY_DATABASE_URL)

import osimport pytest# Устанавливаем `os.environ`, чтобы использовать тестовую БДos.environ['TESTING'] = 'True'from alembic import commandfrom alembic.config import Configfrom app.models import databasefrom sqlalchemy_utils import create_database, drop_database@pytest.fixture(scope="module")def temp_db():    create_database(database.TEST_SQLALCHEMY_DATABASE_URL) # Создаем БД    base_dir = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))    alembic_cfg = Config(os.path.join(base_dir, "alembic.ini")) # Загружаем конфигурацию alembic     command.upgrade(alembic_cfg, "head") # выполняем миграции    try:        yield database.TEST_SQLALCHEMY_DATABASE_URL    finally:        drop_database(database.TEST_SQLALCHEMY_DATABASE_URL) # удаляем БД

def test_sign_up(temp_db):    request_data = {        "email": "vader@deathstar.com",        "name": "Darth Vader",        "password": "rainbow"    }    with TestClient(app) as client:        response = client.post("/sign-up", json=request_data)    assert response.status_code == 200    assert response.json()["id"] == 1    assert response.json()["email"] == "vader@deathstar.com"    assert response.json()["name"] == "Darth"    assert response.json()["token"]["expires"] is not None    assert response.json()["token"]["access_token"] is not None

import asynciofrom app.main import appfrom app.schemas.users import UserCreatefrom app.utils.users import create_user, create_user_tokenfrom fastapi.testclient import TestClientdef test_create_post(temp_db):    user = UserCreate(        email="vader@deathstar.com",        name="Darth",        password="rainbow"    )    request_data = {      "title": "42",      "content": "Don't panic!"    }    with TestClient(app) as client:        # Create user and use his token to add new post        loop = asyncio.get_event_loop()        user_db = loop.run_until_complete(create_user(user))        response = client.post(            "/posts",            json=request_data,            headers={"Authorization": f"Bearer {user_db['token']['token']}"}        )    assert response.status_code == 201    assert response.json()["id"] == 1    assert response.json()["title"] == "42"    assert response.json()["content"] == "Don't panic!"def test_create_post_forbidden_without_token(temp_db):    request_data = {      "title": "42",      "content": "Don't panic!"    }    with TestClient(app) as client:        response = client.post("/posts", json=request_data)    assert response.status_code == 401def test_posts_list(temp_db):    with TestClient(app) as client:        response = client.get("/posts")    assert response.status_code == 200    assert response.json()["total_count"] == 1    assert response.json()["results"][0]["id"] == 1    assert response.json()["results"][0]["title"] == "42"    assert response.json()["results"][0]["content"] == "Don't panic!"def test_post_detail(temp_db):    post_id = 1    with TestClient(app) as client:        response = client.get(f"/posts/{post_id}")    assert response.status_code == 200    assert response.json()["id"] == 1    assert response.json()["title"] == "42"    assert response.json()["content"] == "Don't panic!"def test_update_post(temp_db):    post_id = 1    request_data = {      "title": "42",      "content": "Life? Don't talk to me about life."    }    with TestClient(app) as client:        # Create user token to add new post        loop = asyncio.get_event_loop()        token = loop.run_until_complete(create_user_token(user_id=1))        response = client.put(            f"/posts/{post_id}",            json=request_data,            headers={"Authorization": f"Bearer {token['token']}"}        )    assert response.status_code == 200    assert response.json()["id"] == 1    assert response.json()["title"] == "42"    assert response.json()["content"] == "Life? Don't talk to me about life."def test_update_post_forbidden_without_token(temp_db):    post_id = 1    request_data = {      "title": "42",      "content": "Life? Don't talk to me about life."    }    with TestClient(app) as client:        response = client.put(f"/posts/{post_id}", json=request_data)    assert response.status_code == 401

import asyncioimport pytestfrom app.main import appfrom app.schemas.users import UserCreatefrom app.utils.users import create_user, create_user_tokenfrom fastapi.testclient import TestClientdef test_sign_up(temp_db):    request_data = {        "email": "vader@deathstar.com",        "name": "Darth",        "password": "rainbow"    }    with TestClient(app) as client:        response = client.post("/sign-up", json=request_data)    assert response.status_code == 200    assert response.json()["id"] == 1    assert response.json()["email"] == "vader@deathstar.com"    assert response.json()["name"] == "Darth"    assert response.json()["token"]["expires"] is not None    assert response.json()["token"]["token"] is not Nonedef test_login(temp_db):    request_data = {"username": "vader@deathstar.com", "password": "rainbow"}    with TestClient(app) as client:        response = client.post("/auth", data=request_data)    assert response.status_code == 200    assert response.json()["token_type"] == "bearer"    assert response.json()["expires"] is not None    assert response.json()["access_token"] is not Nonedef test_login_with_invalid_password(temp_db):    request_data = {"username": "vader@deathstar.com", "password": "unicorn"}    with TestClient(app) as client:        response = client.post("/auth", data=request_data)    assert response.status_code == 400    assert response.json()["detail"] == "Incorrect email or password"def test_user_detail(temp_db):    with TestClient(app) as client:        # Create user token to see user info        loop = asyncio.get_event_loop()        token = loop.run_until_complete(create_user_token(user_id=1))        response = client.get(            "/users/me",            headers={"Authorization": f"Bearer {token['token']}"}        )    assert response.status_code == 200    assert response.json()["id"] == 1    assert response.json()["email"] == "vader@deathstar.com"    assert response.json()["name"] == "Darth"def test_user_detail_forbidden_without_token(temp_db):    with TestClient(app) as client:        response = client.get("/users/me")    assert response.status_code == 401@pytest.mark.freeze_time("2015-10-21")def test_user_detail_forbidden_with_expired_token(temp_db, freezer):    user = UserCreate(        email="sidious@deathstar.com",        name="Palpatine",        password="unicorn"    )    with TestClient(app) as client:        # Create user and use expired token        loop = asyncio.get_event_loop()        user_db = loop.run_until_complete(create_user(user))        freezer.move_to("'2015-11-10'")        response = client.get(            "/users/me",            headers={"Authorization": f"Bearer {user_db['token']['token']}"}        )    assert response.status_code == 401

P.S. Исходники