UCarTestTask/main.py at main · arturateo/UCarTestTask · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
import re
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from enum import Enum
from typing import Sequence, Any, Generator, List

from fastapi import FastAPI, Depends
from pydantic import BaseModel, StrictStr
from sqlalchemy import create_engine
from sqlmodel import SQLModel, Field, Session, select


@asynccontextmanager
async def lifespan(app: FastAPI):
    print("INFO:     Создание БД...")
    create_db_and_tables()
    yield
    print("INFO:     Lifespan завершил процесс.")


app = FastAPI(lifespan=lifespan, title="Reviews API")


class SentimentEnum(str, Enum):
    positive = "positive"
    negative = "negative"
    neutral = "neutral"


class ReviewsTable(SQLModel, table=True):
    __table_args__ = {"extend_existing": True}

    id: int | None = Field(default=None, primary_key=True)
    text: str = Field(index=True)
    sentiment: SentimentEnum = Field(index=True)
    created_at: datetime = Field(
        default_factory=lambda: datetime.now(timezone.utc),
        nullable=False,
        index=True,
    )


# 2. ЛУЧШАЯ ПРАКТИКА: Отдельные модели для API
class ReviewsCreateModel(BaseModel):
    text: StrictStr


class ReviewsReadModel(BaseModel):
    id: int
    text: str
    sentiment: SentimentEnum
    created_at: datetime


engine = create_engine("sqlite:///reviews.db")


def create_db_and_tables() -> None:
    SQLModel.metadata.create_all(engine, checkfirst=True)


def get_session() -> Generator[Session, Any, None]:
    with Session(engine) as session:
        yield session


def determine_sentiment(text: str) -> SentimentEnum:
    sentiment_map = {
        "хорош": SentimentEnum.positive,
        "люблю": SentimentEnum.positive,
        "плохо": SentimentEnum.negative,
        "ненавиж": SentimentEnum.negative,
    }

    pattern = r"\b(" + "|".join(sentiment_map.keys()) + r")"
    match = re.search(pattern, text, re.IGNORECASE)

    if match:
        found_word = match.group(1).lower()
        return sentiment_map[found_word]

    return SentimentEnum.neutral


@app.post("/reviews", response_model=ReviewsReadModel)
async def create_reviews(
    review_data: ReviewsCreateModel,
    session: Session = Depends(get_session),
) -> ReviewsTable:
    sentiment_result = determine_sentiment(review_data.text)
    db_review = ReviewsTable(text=review_data.text, sentiment=sentiment_result)

    session.add(db_review)
    session.commit()
    session.refresh(db_review)
    return db_review


@app.get("/reviews", response_model=List[ReviewsReadModel])
async def read_reviews(
    sentiment: SentimentEnum,
    session: Session = Depends(get_session),
) -> Sequence[ReviewsTable]:
    statement = select(ReviewsTable).where(ReviewsTable.sentiment == sentiment)
    results = session.exec(statement).all()
    return results


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("main:app", host="127.0.0.1", port=8000, reload=True)