pythonDev/lr11

//initDB.py
from models import Base, engine, Task
from sqlalchemy.orm import Session
from datetime import date

def init_db():
    Base.metadata.create_all(bind=engine)
    print("Tables created")

    with Session(engine) as session:
        if session.query(Task).count() == 0:
            task1 = Task(
                title="Сделать лабораторную работу",
                description="Прикрепить скриншоты",
                status="pending",
                created_at=date(2025, 4, 1),
                updated_at=date(2025, 4, 2)
            )
            task2 = Task(
                title="Подготовить отчет",
                description="Оформить выводы",
                status="done",
                created_at=date(2025, 4, 3)
            )
            session.add_all([task1, task2])
            session.commit()
            print("Test data has been added")
        else:
            print("Test data already exists")

if __name__ == "__main__":
    init_db()


//-------------------------------------------------------------------------------
// models.py
from sqlalchemy import create_engine, Column, Integer, String, Text, Date
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import date

DATABASE_URL = "sqlite:///tasks.db"
engine = create_engine(DATABASE_URL, echo=False)
Base = declarative_base()
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

class Task(Base):
    __tablename__ = "tasks"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(255), nullable=False)
    description = Column(Text, nullable=True)
    status = Column(String(20), default="pending")
    created_at = Column(Date, default=date.today)
    updated_at = Column(Date, default=date.today, onupdate=date.today)

    def to_dict(self):
        return {
            "id": self.id,
            "title": self.title,
            "description": self.description,
            "status": self.status,
            "created_at": self.created_at.isoformat() if self.created_at else None,
            "updated_at": self.updated_at.isoformat() if self.updated_at else None,
        }

//-------------------------------------------------------------------------------
// app.py
from flask import Flask, request, jsonify
from models import Task, SessionLocal, engine
from sqlalchemy.orm import Session
from sqlalchemy import or_
from datetime import date

app = Flask(__name__)

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

@app.route('/tasks', methods=['GET'])
def get_tasks():
    with Session(engine) as db:
        query = db.query(Task)

        if request.args.get('sort') == 'created_at':
            query = query.order_by(Task.created_at)

        q = request.args.get('q')
        if q:
            query = query.filter(Task.title.contains(q))

        tasks = query.all()
        return jsonify([task.to_dict() for task in tasks])

@app.route('/tasks/search', methods=['GET'])
def search_tasks():
    q = request.args.get('q')
    if not q:
        return jsonify([])

    with Session(engine) as db:
        tasks = db.query(Task).filter(Task.title.contains(q)).all()
        return jsonify([task.to_dict() for task in tasks])

@app.route('/tasks', methods=['POST'])
def create_task():
    data = request.get_json()
    if not data or not data.get('title'):
        return jsonify({"error": "title is required"}), 400

    new_task = Task(
        title=data['title'],
        description=data.get('description'),
        status=data.get('status', 'pending')
    )

    with Session(engine) as db:
        db.add(new_task)
        db.commit()
        db.refresh(new_task)
        return jsonify(new_task.to_dict()), 201

@app.route('/tasks/<int:id>', methods=['GET'])
def get_task(id):
    with Session(engine) as db:
        task = db.query(Task).filter(Task.id == id).first()
        if not task:
            return jsonify({"error": "Task not found"}), 404
        return jsonify(task.to_dict())

@app.route('/tasks/<int:id>', methods=['PUT'])
def update_task(id):
    data = request.get_json()
    if not data:
        return jsonify({"error": "No data provided"}), 400

    with Session(engine) as db:
        task = db.query(Task).filter(Task.id == id).first()
        if not task:
            return jsonify({"error": "Task not found"}), 404

        task.title = data.get('title', task.title)
        task.description = data.get('description', task.description)
        task.status = data.get('status', task.status)
        task.updated_at = date.today()

        db.commit()
        db.refresh(task)
        return jsonify(task.to_dict())

@app.route('/tasks/<int:id>', methods=['DELETE'])
def delete_task(id):
    with Session(engine) as db:
        task = db.query(Task).filter(Task.id == id).first()
        if not task:
            return jsonify({"error": "Task not found"}), 404

        db.delete(task)
        db.commit()
        return '', 204

if __name__ == '__main__':
    app.run(debug=True)


//-------------------------------------------------------------------------------
// test_api.py
import requests

BASE_URL = "http://127.0.0.1:5000"

def test_crud():

    # 1. Создание задачи
    print("1. Тест: создание задачи")
    try:
        resp = requests.post(f"{BASE_URL}/tasks", json={"title": "Авто-тестовая задача", "description": "Тест через requests"})
        assert resp.status_code == 201, f"Ожидался статус 201, получен {resp.status_code}"
        task_data = resp.json()
        task_id = task_data["id"]
        print(f"Успешно создана задача с ID = {task_id}")
    except Exception as e:
        print(f"Ошибка при создании задачи: {e}")
        return

    # 2. Получение задачи по ID
    print("\n2. Тест: получение задачи по ID")
    try:
        resp = requests.get(f"{BASE_URL}/tasks/{task_id}")
        assert resp.status_code == 200, f"Ожидался статус 200, получен {resp.status_code}"
        assert resp.json()["title"] == "Авто-тестовая задача"
        print(f"Задача с ID {task_id} успешно получена")
    except Exception as e:
        print(f"Ошибка при получении задачи: {e}")
        return

    # 3. Обновление задачи
    print("\n3. Тест: обновление задачи")
    try:
        update_data = {"title": "Обновлённая авто-задача", "status": "done"}
        resp = requests.put(f"{BASE_URL}/tasks/{task_id}", json=update_data)
        assert resp.status_code == 200, f"Ожидался статус 200, получен {resp.status_code}"
        updated = resp.json()
        assert updated["title"] == "Обновлённая авто-задача"
        assert updated["status"] == "done"
        print(f"Задача с ID {task_id} успешно обновлена")
    except Exception as e:
        print(f"Ошибка при обновлении задачи: {e}")
        return

    # 4. Удаление задачи
    print("\n4. Тест: удаление задачи")
    try:
        resp = requests.delete(f"{BASE_URL}/tasks/{task_id}")
        assert resp.status_code == 204, f"Ожидался статус 204, получен {resp.status_code}"
        print(f"Задача с ID {task_id} успешно удалена")
    except Exception as e:
        print(f"Ошибка при удалении задачи: {e}")
        return

    # 5. Проверка, что задача действительно удалена
    print("\n5. Тест: проверка отсутствия удалённой задачи")
    try:
        resp = requests.get(f"{BASE_URL}/tasks/{task_id}")
        assert resp.status_code == 404, f"Ожидался статус 404, получен {resp.status_code}"
        print(f"Задача с ID {task_id} действительно отсутствует (404 получен)")
    except Exception as e:
        print(f"Ошибка при проверке удаления: {e}")
        return

    # 6. Тест поиска
    print("\n6. Тест: поиск задач по заголовку (/tasks?q=...)")
    try:
        resp = requests.get(f"{BASE_URL}/tasks?q=лабораторную")
        assert resp.status_code == 200
        tasks = resp.json()
        assert len(tasks) >= 1
        print(f"Поиск найден: {len(tasks)} задач(а) по ключевому слову 'лабораторную'")
    except Exception as e:
        print(f"Ошибка при поиске: {e}")

    # 7. Тест сортировки
    print("\n7. Тест: сортировка по дате создания (/tasks?sort=created_at)")
    try:
        resp = requests.get(f"{BASE_URL}/tasks?sort=created_at")
        assert resp.status_code == 200
        tasks = resp.json()
        dates = [t["created_at"] for t in tasks]
        if dates == sorted(dates):
            print(f"Сортировка по created_at работает корректно")
        else:
            print(f"Сортировка работает некорректно")
    except Exception as e:
        print(f"Ошибка при проверке сортировки: {e}")


if __name__ == "__main__":
    try:
        test_crud()
    except requests.exceptions.ConnectionError:
        print("Не удаётся подключиться к серверу")
    except Exception as e:
        print(f"Неожиданная ошибка: {e}")