done

README.MD

@@ -0,0 +1,76 @@
+# Проект CRUD API для пользователей и постов
+
+Это FastAPI приложение предоставляет REST API для управления пользователями и постами с использованием базы данных PostgreSQL.
+
+## Эндпоинты
+
+### Пользователи
+
+- **POST /users/**  
+  Создает нового пользователя на основе предоставленных данных.  
+  Тело запроса: `{"name": "string", "email": "string", "password": "string"}`  
+  Ответ: `{"id": int, "name": "string", "email": "string"}`
+
+- **GET /users/**  
+  Возвращает список всех пользователей.
+
+- **GET /users/{user_id}**  
+  Возвращает информацию о пользователе по его ID.
+
+- **PUT /users/{user_id}**  
+  Обновляет данные пользователя по его ID.  
+  Тело запроса: `{"name": "string", "email": "string", "password": "string"}`
+
+- **DELETE /users/{user_id}**  
+  Удаляет пользователя по его ID.
+
+### Посты
+
+- **POST /posts/**  
+  Создает новый пост на основе предоставленных данных.  
+  Тело запроса: `{"title": "string", "content": "string", "user_id": int}`  
+  Ответ: `{"id": int, "title": "string", "content": "string", "user_id": int}`
+
+- **GET /posts/**  
+  Возвращает список всех постов.
+
+- **GET /posts/{post_id}**  
+  Возвращает информацию о посте по его ID.
+
+- **PUT /posts/{post_id}**  
+  Обновляет данные поста по его ID.  
+  Тело запроса: `{"title": "string", "content": "string", "user_id": int}`
+
+## Запуск
+
+1. Установите зависимости: `pip install fastapi sqlalchemy psycopg2-binary uvicorn`
+2. Настройте базу данных в `database.py`
+3. Запустите приложение: `uvicorn main:app --reload`
+
+## Безопасность и журналирование
+
+- **Хэширование паролей**: Пароли пользователей хэшируются с использованием bcrypt.
+- **Шифрование данных**: Email пользователей шифруется в базе данных с использованием Fernet (AES).
+- **Защита от SQL-инъекций**: Все запросы параметризованы через SQLAlchemy.
+- **Журналирование**: Действия пользователей логируются (создание, обновление и т.д.).
+
+## Структура проекта
+
+- `main.py`: Основное приложение FastAPI с эндпоинтами и журналированием
+- `crud.py`: Функции для работы с базой данных
+- `models.py`: Модели SQLAlchemy для таблиц
+- `schemas.py`: Pydantic схемы для валидации данных
+- `database.py`: Настройки подключения к базе данных
+- `utils.py`: Утилиты для хэширования и шифрования
+- `test_main.py`: Тесты для API и базы данных
+
+## Тестирование
+
+Для запуска тестов установите pytest: `pip install pytest`
+
+Запустите тесты: `pytest test_main.py`
+
+Тесты включают:
+- Проверку доступности базы данных
+- CRUD операции для пользователей и постов
+- Очистку тестовой базы данных

crud.py

@@ -0,0 +1,139 @@
+from sqlalchemy.orm import Session
+from sqlalchemy.exc import SQLAlchemyError, IntegrityError
+import models, schemas
+from utils import hash_password, encrypt_data, decrypt_data
+
+
+def create_user(db: Session, user: schemas.UserCreate):
+    """Создает нового пользователя в базе данных на основе предоставленных данных."""
+    try:
+        hashed_password = hash_password(user.password)
+        encrypted_email = encrypt_data(user.email)
+        db_user = models.User(name=user.name, email=encrypted_email, password=hashed_password)
+        db.add(db_user)
+        db.commit()
+        db.refresh(db_user)
+        db_user.email = decrypt_data(db_user.email)
+        return db_user
+    except IntegrityError:
+        db.rollback()
+        raise ValueError("User with this email already exists")
+    except SQLAlchemyError:
+        db.rollback()
+        raise ConnectionError("Database error while creating user")
+
+
+def get_users(db: Session):
+    """Получает список всех пользователей из базы данных."""
+    try:
+        users = db.query(models.User).all()
+        for user in users:
+            user.email = decrypt_data(user.email)
+        return users
+    except SQLAlchemyError:
+        raise ConnectionError("Database error while fetching users")
+
+
+def get_user(db: Session, user_id: int):
+    """Получает информацию о пользователе по его ID из базы данных."""
+    try:
+        user = db.query(models.User).filter(models.User.id == user_id).first()
+        if not user:
+            raise LookupError("User not found")
+        user.email = decrypt_data(user.email)
+        return user
+    except SQLAlchemyError:
+        raise ConnectionError("Database error while fetching user")
+
+
+def delete_user(db: Session, user_id: int):
+    """Удаляет пользователя по его ID из базы данных."""
+    try:
+        user = db.query(models.User).filter(models.User.id == user_id).first()
+        if not user:
+            raise LookupError("User not found")
+
+        db.delete(user)
+        db.commit()
+        return user
+    except SQLAlchemyError:
+        db.rollback()
+        raise ConnectionError("Database error while deleting user")
+
+
+def update_user(db: Session, user_id: int, user_update: schemas.UserCreate):
+    """Обновляет данные пользователя по его ID."""
+    try:
+        user = db.query(models.User).filter(models.User.id == user_id).first()
+        if not user:
+            raise LookupError("User not found")
+
+        user.name = user_update.name
+        user.email = encrypt_data(user_update.email)
+        user.password = hash_password(user_update.password)
+
+        db.commit()
+        db.refresh(user)
+        user.email = decrypt_data(user.email)
+        return user
+    except SQLAlchemyError:
+        db.rollback()
+        raise ConnectionError("Database error while updating user")
+
+
+def create_post(db: Session, post: schemas.PostCreate):
+    """Создает новый пост в базе данных, проверяя существование пользователя."""
+    try:
+        user = db.query(models.User).filter(models.User.id == post.user_id).first()
+        if not user:
+            raise LookupError("User does not exist")
+
+        db_post = models.Post(**post.dict())
+        db.add(db_post)
+        db.commit()
+        db.refresh(db_post)
+        return db_post
+
+    except IntegrityError:
+        db.rollback()
+        raise ValueError("Invalid post data")
+    except SQLAlchemyError:
+        db.rollback()
+        raise ConnectionError("Database error while creating post")
+
+
+def get_posts(db: Session):
+    """Получает список всех постов из базы данных."""
+    try:
+        return db.query(models.Post).all()
+    except SQLAlchemyError:
+        raise ConnectionError("Database error while fetching posts")
+
+
+def get_post(db: Session, post_id: int):
+    """Получает информацию о посте по его ID из базы данных."""
+    try:
+        post = db.query(models.Post).filter(models.Post.id == post_id).first()
+        if not post:
+            raise LookupError("Post not found")
+        return post
+    except SQLAlchemyError:
+        raise ConnectionError("Database error while fetching post")
+
+
+def update_post(db: Session, post_id: int, post_update: schemas.PostCreate):
+    """Обновляет данные поста по его ID."""
+    try:
+        post = db.query(models.Post).filter(models.Post.id == post_id).first()
+        if not post:
+            raise LookupError("Post not found")
+
+        for key, value in post_update.dict().items():
+            setattr(post, key, value)
+
+        db.commit()
+        db.refresh(post)
+        return post
+    except SQLAlchemyError:
+        db.rollback()
+        raise ConnectionError("Database error while updating post")

database.py

@@ -0,0 +1,9 @@
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker, declarative_base
+
+DATABASE_URL = "postgresql://postgres:postgres@localhost:5432/testdb"
+
+engine = create_engine(DATABASE_URL)
+SessionLocal = sessionmaker(bind=engine)
+
+Base = declarative_base()

main.py

@@ -0,0 +1,122 @@
+from fastapi import FastAPI, Depends, HTTPException
+from sqlalchemy.orm import Session
+import logging
+import models, schemas, crud
+from database import SessionLocal, engine
+
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+models.Base.metadata.create_all(bind=engine)
+
+app = FastAPI()
+
+
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+
+
+@app.post("/users/", response_model=schemas.UserResponse)
+def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
+    """Создает нового пользователя на основе предоставленных данных."""
+    try:
+        result = crud.create_user(db, user)
+        logger.info(f"User created: {result.id}")
+        return result
+    except ValueError as e:
+        logger.warning(f"Failed to create user: {e}")
+        raise HTTPException(status_code=400, detail=str(e))
+    except ConnectionError as e:
+        logger.error(f"Database error: {e}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.get("/users/")
+def read_users(db: Session = Depends(get_db)):
+    """Возвращает список всех пользователей."""
+    try:
+        return crud.get_users(db)
+    except ConnectionError as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.get("/users/{user_id}")
+def read_user(user_id: int, db: Session = Depends(get_db)):
+    """Возвращает информацию о пользователе по его ID."""
+    try:
+        return crud.get_user(db, user_id)
+    except LookupError as e:
+        raise HTTPException(status_code=404, detail=str(e))
+    except ConnectionError as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.delete("/users/{user_id}")
+def delete_user(user_id: int, db: Session = Depends(get_db)):
+    """Удаляет пользователя по его ID."""
+    try:
+        return crud.delete_user(db, user_id)
+    except LookupError as e:
+        raise HTTPException(status_code=404, detail=str(e))
+    except ConnectionError as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.put("/users/{user_id}", response_model=schemas.UserResponse)
+def update_user(user_id: int, user: schemas.UserCreate, db: Session = Depends(get_db)):
+    """Обновляет данные пользователя по его ID."""
+    try:
+        return crud.update_user(db, user_id, user)
+    except LookupError as e:
+        raise HTTPException(status_code=404, detail=str(e))
+    except ConnectionError as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.post("/posts/")
+def create_post(post: schemas.PostCreate, db: Session = Depends(get_db)):
+    """Создает новый пост на основе предоставленных данных."""
+    try:
+        return crud.create_post(db, post)
+    except LookupError as e:
+        raise HTTPException(status_code=404, detail=str(e))
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+    except ConnectionError as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.get("/posts/")
+def read_posts(db: Session = Depends(get_db)):
+    """Возвращает список всех постов."""
+    try:
+        return crud.get_posts(db)
+    except ConnectionError as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.get("/posts/{post_id}")
+def read_post(post_id: int, db: Session = Depends(get_db)):
+    """Возвращает информацию о посте по его ID."""
+    try:
+        return crud.get_post(db, post_id)
+    except LookupError as e:
+        raise HTTPException(status_code=404, detail=str(e))
+    except ConnectionError as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.put("/posts/{post_id}", response_model=schemas.PostResponse)
+def update_post(post_id: int, post: schemas.PostCreate, db: Session = Depends(get_db)):
+    """Обновляет данные поста по его ID."""
+    try:
+        return crud.update_post(db, post_id, post)
+    except LookupError as e:
+        raise HTTPException(status_code=404, detail=str(e))
+    except ConnectionError as e:
+        raise HTTPException(status_code=500, detail=str(e))

models.py

@@ -0,0 +1,26 @@
+from sqlalchemy import Column, Integer, String, ForeignKey
+from sqlalchemy.orm import relationship
+from database import Base
+
+
+class User(Base):
+    __tablename__ = "users"
+
+    id = Column(Integer, primary_key=True, index=True)
+    name = Column(String)
+    email = Column(String)
+    password = Column(String)
+
+    posts = relationship("Post", back_populates="owner")
+
+
+class Post(Base):
+    __tablename__ = "posts"
+
+    id = Column(Integer, primary_key=True, index=True)
+    title = Column(String)
+    content = Column(String)
+
+    user_id = Column(Integer, ForeignKey("users.id"))
+
+    owner = relationship("User", back_populates="posts")

schemas.py

@@ -0,0 +1,23 @@
+from pydantic import BaseModel, ConfigDict
+
+
+class UserCreate(BaseModel):
+    name: str
+    email: str
+    password: str
+
+class UserResponse(BaseModel):
+    id: int
+    name: str
+    email: str
+    model_config = ConfigDict(from_attributes=True)
+
+
+class PostCreate(BaseModel):
+    title: str
+    content: str
+    user_id: int
+
+class PostResponse(PostCreate):
+    id: int
+    model_config = ConfigDict(from_attributes=True)

test_main.py

@@ -0,0 +1,172 @@
+import pytest
+from sqlalchemy import create_engine, text
+from sqlalchemy.orm import sessionmaker
+from fastapi.testclient import TestClient
+
+import models, schemas
+from main import app, get_db
+from database import engine, Base
+
+# Используем основную базу данных для тестов
+TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+def override_get_db():
+    db = TestingSessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+
+app.dependency_overrides[get_db] = override_get_db
+
+client = TestClient(app)
+
+@pytest.fixture(scope="session", autouse=True)
+def setup_database():
+    # Удаляем и создаем таблицы заново для учета изменений схемы
+    Base.metadata.drop_all(bind=engine)
+    Base.metadata.create_all(bind=engine)
+    yield
+
+@pytest.fixture(scope="function", autouse=True)
+def clean_database():
+    """Очистка записей в базе данных перед каждым тестом."""
+    db = TestingSessionLocal()
+    try:
+        # Удаляем все записи из таблиц
+        db.execute(text("DELETE FROM posts"))
+        db.execute(text("DELETE FROM users"))
+        db.commit()
+    finally:
+        db.close()
+    yield
+
+# 1.5.1 Тесты до начала основного сценария (проверка доступности базы данных)
+def test_database_connection():
+    """Проверяет доступность базы данных."""
+    db = TestingSessionLocal()
+    try:
+        # Простой запрос для проверки соединения
+        result = db.execute(text("SELECT 1"))
+        assert result.fetchone()[0] == 1
+    finally:
+        db.close()
+
+# 1.5.2 Основные тесты для CRUD операций
+
+# Пользователи
+def test_create_user():
+    """Проверяет добавление новой записи в таблицу пользователей."""
+    user_data = {"name": "Test User", "email": "test@example.com", "password": "password123"}
+    response = client.post("/users/", json=user_data)
+    assert response.status_code == 200
+    data = response.json()
+    assert data["name"] == user_data["name"]
+    assert data["email"] == user_data["email"]
+    assert "id" in data
+
+def test_get_users():
+    """Проверяет просмотр всех записей пользователей."""
+    response = client.get("/users/")
+    assert response.status_code == 200
+    data = response.json()
+    assert isinstance(data, list)
+
+def test_get_user():
+    """Проверяет просмотр конкретного пользователя."""
+    # Сначала создадим пользователя
+    user_data = {"name": "Another User", "email": "another@example.com", "password": "pass456"}
+    create_response = client.post("/users/", json=user_data)
+    user_id = create_response.json()["id"]
+
+    response = client.get(f"/users/{user_id}")
+    assert response.status_code == 200
+    data = response.json()
+    assert data["id"] == user_id
+    assert data["name"] == user_data["name"]
+
+def test_update_user():
+    """Проверяет редактирование данных пользователя."""
+    # Создаем пользователя
+    user_data = {"name": "Update User", "email": "update@example.com", "password": "oldpass"}
+    create_response = client.post("/users/", json=user_data)
+    user_id = create_response.json()["id"]
+
+    # Обновляем
+    update_data = {"name": "Updated User", "email": "updated@example.com", "password": "newpass"}
+    response = client.put(f"/users/{user_id}", json=update_data)
+    assert response.status_code == 200
+    data = response.json()
+    assert data["name"] == update_data["name"]
+    assert data["email"] == update_data["email"]
+
+def test_delete_user():
+    """Проверяет удаление записи пользователя."""
+    # Создаем пользователя
+    user_data = {"name": "Delete User", "email": "delete@example.com", "password": "delpass"}
+    create_response = client.post("/users/", json=user_data)
+    user_id = create_response.json()["id"]
+
+    # Удаляем
+    response = client.delete(f"/users/{user_id}")
+    assert response.status_code == 200
+
+    # Проверяем, что больше не существует
+    response = client.get(f"/users/{user_id}")
+    assert response.status_code == 404
+
+# Посты
+def test_create_post():
+    """Проверяет добавление новой записи в таблицу постов."""
+    user_data = {"name": "Post User", "email": "post@example.com", "password": "postpass"}
+    user_response = client.post("/users/", json=user_data)
+    user_id = user_response.json()["id"]
+
+    post_data = {"title": "Test Post", "content": "Test content", "user_id": user_id}
+    response = client.post("/posts/", json=post_data)
+    assert response.status_code == 200
+    data = response.json()
+    assert data["title"] == post_data["title"]
+    assert data["content"] == post_data["content"]
+    assert data["user_id"] == user_id
+
+def test_get_posts():
+    """Проверяет просмотр всех записей постов."""
+    response = client.get("/posts/")
+    assert response.status_code == 200
+    data = response.json()
+    assert isinstance(data, list)
+
+def test_get_post():
+    """Проверяет просмотр конкретного поста."""
+    user_data = {"name": "Post User 2", "email": "post2@example.com", "password": "post2pass"}
+    user_response = client.post("/users/", json=user_data)
+    user_id = user_response.json()["id"]
+
+    post_data = {"title": "Another Post", "content": "Another content", "user_id": user_id}
+    create_response = client.post("/posts/", json=post_data)
+    post_id = create_response.json()["id"]
+
+    response = client.get(f"/posts/{post_id}")
+    assert response.status_code == 200
+    data = response.json()
+    assert data["id"] == post_id
+    assert data["title"] == post_data["title"]
+
+def test_update_post():
+    """Проверяет редактирование данных поста."""
+    user_data = {"name": "Update Post User", "email": "updatepost@example.com", "password": "updatepostpass"}
+    user_response = client.post("/users/", json=user_data)
+    user_id = user_response.json()["id"]
+
+    post_data = {"title": "Update Post", "content": "Update content", "user_id": user_id}
+    create_response = client.post("/posts/", json=post_data)
+    post_id = create_response.json()["id"]
+
+    # Обновляем
+    update_data = {"title": "Updated Post", "content": "Updated content", "user_id": user_id}
+    response = client.put(f"/posts/{post_id}", json=update_data)
+    assert response.status_code == 200
+    data = response.json()
+    assert data["title"] == update_data["title"]
+    assert data["content"] == update_data["content"]

uml_diagram.puml

@@ -0,0 +1,57 @@
+@startuml UML Class Diagram
+
+class User {
+    +id: Integer (primary key)
+    +name: String
+    +email: String (encrypted)
+    +password: String (hashed)
+    +posts: Relationship<Post>
+    --
+    +create_user(db, user)
+    +get_users(db)
+    +get_user(db, user_id)
+    +update_user(db, user_id, user_update)
+    +delete_user(db, user_id)
+}
+
+class Post {
+    +id: Integer (primary key)
+    +title: String
+    +content: String
+    +user_id: Integer (foreign key)
+    +owner: Relationship<User>
+    --
+    +create_post(db, post)
+    +get_posts(db)
+    +get_post(db, post_id)
+    +update_post(db, post_id, post_update)
+}
+
+class UserCreate {
+    +name: String
+    +email: String
+    +password: String
+}
+
+class UserResponse {
+    +id: Integer
+    +name: String
+    +email: String
+}
+
+class PostCreate {
+    +title: String
+    +content: String
+    +user_id: Integer
+}
+
+class PostResponse {
+    +id: Integer
+    +title: String
+    +content: String
+    +user_id: Integer
+}
+
+User ||--o{ Post : "has many"
+
+@enduml

utils.py

@@ -0,0 +1,23 @@
+from passlib.context import CryptContext
+from cryptography.fernet import Fernet
+import os
+
+# Хэширование паролей
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+def hash_password(password: str) -> str:
+    return pwd_context.hash(password)
+
+def verify_password(plain_password: str, hashed_password: str) -> bool:
+    return pwd_context.verify(plain_password, hashed_password)
+
+# Шифрование (например, для email)
+# Генерируем ключ, в продакшене хранить securely
+key = os.getenv("ENCRYPTION_KEY", Fernet.generate_key())
+cipher = Fernet(key)
+
+def encrypt_data(data: str) -> str:
+    return cipher.encrypt(data.encode()).decode()
+
+def decrypt_data(encrypted_data: str) -> str:
+    return cipher.decrypt(encrypted_data.encode()).decode()