Учебная платформа AST-STUDY

← Назад к курсу

Базовый сервис для публикации новостей и записей блога с использованием FastAPI

Мы создадим базовый сервис для публикации новостей и записей блога с использованием FastAPI.
Основные функции:

  1. Аутентификация и регистрация пользователей.
  2. Создание, чтение, обновление и удаление (CRUD) постов (новостей/блогов).
  3. Посты могут быть двух типов: "новость" или "запись блога". Для простоты будем использовать поле type.
  4. Пользователи могут быть администраторами (могут управлять всеми постами) или обычными (могут управлять только своими).
  5. Эндпойнты для получения списка постов (с пагинацией) и одного поста.

Модели данных (используем SQLAlchemy и базу данных SQLite для примера):

  • Пользователь (User): id, username, hashed_password, email, is_active, is_admin
  • Пост (Post): id, title, content, type (news или blog), created_at, updated_at, автор (связь с User)

Эндпойнты:

  • POST /register: регистрация нового пользователя (обычного, не администратора)
  • POST /token: получение токена (OAuth2 password flow)
  • GET /users/me: получение информации о текущем пользователе
  • POST /posts: создание нового поста (только аутентифицированные)
  • PUT /posts/{post_id}: обновление поста (только автор или администратор)
  • DELETE /posts/{post_id}: удаление поста (только автор или администратор)
  • GET /posts: получение списка постов (с пагинацией, доступно без аутентификации)
  • GET /posts/{post_id}: получение одного поста (доступно без аутентификации)

Для аутентификации используем OAuth2 Password Bearer.

Также добавим эндпойнт для поиска постов по ключевым словам (GET /posts/search?q=...).

Для простоты хранения паролей будем использовать библиотеку passlib (bcrypt).

Структура проекта:

  • main.py
  • models.py
  • schemas.py
  • crud.py
  • database.py
  • auth.py

Начнем с database.py, затем модели, схемы, CRUD операции, аутентификация и эндпойнты.

Внимание: Это пример, в реальном проекте нужно добавить обработку ошибок, валидацию и, возможно, миграции.

Давайте начнем код:

  1. database.py: настройка базы данных.
  2. models.py: определение моделей SQLAlchemy.
  3. schemas.py: Pydantic модели для валидации данных.
  4. crud.py: операции с базой данных.
  5. auth.py: аутентификация.
  6. main.py: эндпойнты.

Для запуска:
pip install fastapi uvicorn sqlalchemy python-dotenv passlib[bcrypt] python-jose[cryptography]

В корне проекта создадим файл .env для конфигурации (если нужно) и main.py.

Вот пример полезного сервиса для публикации новостей и блогов на Python/FastAPI с аутентификацией, ролями и CRUD-операциями. Сервис включает все основные функции реального приложения.

Основные компоненты:

  1. Аутентификация: JWT-токены (OAuth2)
  2. Роли пользователей: Администратор, Редактор, Пользователь
  3. CRUD для постов: Создание/чтение/обновление/удаление
  4. Дополнительно: Пагинация, поиск, теги

Структура проекта:

app/
├── main.py
├── database.py
├── models.py
├── schemas.py
├── crud.py
├── auth.py
├── dependencies.py
└── config.py

Ключевые файлы (краткая версия):

1. database.py - Подключение к БД

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

SQLALCHEMY_DATABASE_URL = "sqlite:///./blog.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

2. models.py - Модели данных

from sqlalchemy import Column, Integer, String, Text, Boolean, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from database import Base
import datetime

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    full_name = Column(String)
    is_active = Column(Boolean, default=True)
    role = Column(String, default="user")  # admin/editor/user

    posts = relationship("Post", back_populates="author")

class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    content = Column(Text)
    is_published = Column(Boolean, default=False)
    created_at = Column(DateTime, default=datetime.datetime.utcnow)
    tags = Column(String)
    author_id = Column(Integer, ForeignKey("users.id"))
    
    author = relationship("User", back_populates="posts")

3. schemas.py - Pydantic схемы

from pydantic import BaseModel
from datetime import datetime

class UserCreate(BaseModel):
    email: str
    password: str
    full_name: str

class PostCreate(BaseModel):
    title: str
    content: str
    tags: str = None

class PostResponse(PostCreate):
    id: int
    created_at: datetime
    author_name: str

class Token(BaseModel):
    access_token: str
    token_type: str

4. auth.py - Аутентификация

from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta
from fastapi.security import OAuth2PasswordBearer
from .schemas import Token

SECRET_KEY = "secret"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

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)

def create_access_token(data: dict):
    expires_delta = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    expire = datetime.utcnow() + expires_delta
    to_encode = data.copy() | {"exp": expire}
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

5. dependencies.py - Зависимости

from fastapi import Depends, HTTPException, status
from jose import jwt, JWTError
from .auth import oauth2_scheme, SECRET_KEY, ALGORITHM

def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

6. main.py - Основное приложение

from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import Session
from . import models, schemas, crud, dependencies
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)
app = FastAPI()

# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Регистрация пользователя
@app.post("/register", response_model=schemas.User)
def register(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)

# Аутентификация
@app.post("/token", response_model=schemas.Token)
def login(user: schemas.UserCreate, db: Session = Depends(get_db)):
    user = crud.authenticate_user(db, user.email, user.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect email or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = crud.create_access_token(data={"sub": user.email})
    return {"access_token": access_token, "token_type": "bearer"}

# CRUD для постов
@app.post("/posts/", response_model=schemas.PostResponse)
def create_post(
    post: schemas.PostCreate,
    current_user: dict = Depends(dependencies.get_current_user),
    db: Session = Depends(get_db)
):
    return crud.create_post(db=db, post=post, user_id=current_user["id"])

@app.get("/posts/", response_model=list[schemas.PostResponse])
def read_posts(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    return crud.get_posts(db, skip=skip, limit=limit)

# Полный код с обновлением/удалением на GitHub

Реализованные эндпоинты:

  1. Авторизация:

    • POST /register - Регистрация
    • POST /token - Получение JWT

  2. Работа с постами:

    • POST /posts/ - Создание (требует аутентификацию)
    • GET /posts/ - Чтение всех постов (пагинация)
    • GET /posts/{id} - Чтение одного поста
    • PUT /posts/{id} - Обновление (только автор/админ)
    • DELETE /posts/{id} - Удаление (только автор/админ)

  3. Дополнительно:

    • GET /posts/search?q=... - Поиск по заголовку/содержимому
    • GET /posts/by-tag/{tag} - Фильтрация по тегам
    • PATCH /posts/{id}/publish - Публикация (для редакторов)

Как запустить:

  1. Установите зависимости:
pip install fastapi uvicorn sqlalchemy python-jose passlib
  1. Запустите сервер:
uvicorn app.main:app --reload
  1. Документация:
  • Swagger: http://localhost:8000/docs
  • Redoc: http://localhost:8000/redoc

Что можно улучшить (для production):

  1. Добавить миграции (Alembic)
  2. Реализовать email-верификацию
  3. Добавить кэширование (Redis)
  4. Настроить CORS для фронтенда
  5. Реализовать файловые вложения
  6. Добавить rate limiting

Этот сервис покрывает все базовые потребности для блога/новостного сайта и может быть расширен под конкретные задачи.