init

README.MD

@@ -1,80 +0,0 @@
-# Проект 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. При необходимости задайте постоянный ключ шифрования: `set ENCRYPTION_KEY=<key>` (Windows PowerShell) или `export ENCRYPTION_KEY=<key>` (Linux/macOS)
-4. Запустите приложение: `uvicorn main:app --reload`
-
-> По умолчанию приложение сохранит сгенерированный ключ в файле `encryption.key` рядом с `utils.py`.
-> Если этот файл удалён или ключ изменён, ранее зашифрованные email-адреса перестанут раскрываться.
-
-## Безопасность и журналирование
-
-- **Хэширование паролей**: Пароли пользователей хэшируются с использованием 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

@@ -1,139 +0,0 @@
-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

@@ -1,9 +0,0 @@
-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()

encryption.key

@@ -1 +0,0 @@
-05DWJYSMMsTlG0bzGBOPDRLopKVXYo4rQmcJD8TAn3U=

main.py

@@ -1,122 +0,0 @@
-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

@@ -1,26 +0,0 @@
-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")

requirements.txt

@@ -1,8 +0,0 @@
-cryptography==48.0.0
-fastapi==0.136.1
-passlib==1.7.4
-pydantic==2.13.4
-pytest==9.0.3
-SQLAlchemy==2.0.49
-psycopg2
-uvicorn==0.34.0

schemas.py

@@ -1,23 +0,0 @@
-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

@@ -1,162 +0,0 @@
-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
-
-
-def test_database_connection():
-    """Проверяет доступность базы данных."""
-    db = TestingSessionLocal()
-    try:
-        # Простой запрос для проверки соединения
-        result = db.execute(text("SELECT 1"))
-        assert result.fetchone()[0] == 1
-    finally:
-        db.close()
-
-
-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

@@ -1,57 +0,0 @@
-@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

@@ -1,48 +0,0 @@
-from passlib.context import CryptContext
-from cryptography.fernet import Fernet, InvalidToken
-from pathlib import Path
-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)
-
-
-KEY_FILE = Path(__file__).resolve().parent / "encryption.key"
-
-
-def load_key() -> bytes:
-    env_key = os.getenv("ENCRYPTION_KEY")
-    if env_key:
-        return env_key.encode() if isinstance(env_key, str) else env_key
-
-    if KEY_FILE.exists():
-        return KEY_FILE.read_bytes().strip()
-
-    key = Fernet.generate_key()
-    KEY_FILE.write_bytes(key)
-    return key
-
-
-try:
-    key = load_key()
-    cipher = Fernet(key)
-except Exception as exc:
-    raise RuntimeError(
-        "Invalid encryption key. Set ENCRYPTION_KEY or remove/replace encryption.key with a valid Fernet key."
-    ) from exc
-
-
-def encrypt_data(data: str) -> str:
-    return cipher.encrypt(data.encode()).decode()
-
-def decrypt_data(encrypted_data: str) -> str:
-    try:
-        return cipher.decrypt(encrypted_data.encode()).decode()
-    except InvalidToken as exc:
-        raise ValueError("Unable to decrypt data: invalid encryption token or mismatched key") from exc