FastApi

1
from datetime import datetime
2
from enum import Enum
3
from typing import List, Optional
4
from pydantic import BaseModel, Field
5

6
from fastapi import FastAPI, Request, status
7
from fastapi.encoders import jsonable_encoder
8
from fastapi.exceptions import ValidationError
9
from fastapi.responses import JSONResponse
10

11
app = FastAPI(
12
    title="Trading App"
13
)
14

15

16
# Благодаря этой функции клиент видит ошибки, происходящие на сервере, вместо "Internal server error"
17
@app.exception_handler(ValidationError)
18
async def validation_exception_handler(request: Request, exc: ValidationError):
19
    return JSONResponse(
20
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
21
        content=jsonable_encoder({"detail": exc.errors()}),
22
    )
23

24

25
fake_users = [
26
    {"id": 1, "role": "admin", "name": ["Bob"]},
27
    {"id": 2, "role": "investor", "name": "John"},
28
    {"id": 3, "role": "trader", "name": "Matt"},
29
    {"id": 4, "role": "investor", "name": "Homer", "degree": [
30
        {"id": 1, "created_at": "2020-01-01T00:00:00", "type_degree": "expert"}
31
    ]},
32
]
33

34

35
class DegreeType(Enum):
36
    newbie = "newbie"
37
    expert = "expert"
38

39

40
class Degree(BaseModel):
41
    id: int
42
    created_at: datetime
43
    type_degree: DegreeType
44

45

46
class User(BaseModel):
47
    id: int
48
    role: str
49
    name: str
50
    degree: Optional[List[Degree]] = []
51

52

53
@app.get("/users/{user_id}", response_model=List[User])
54
def get_user(user_id: int):
55
    return [user for user in fake_users if user.get("id") == user_id]
56

57

58
fake_trades = [
59
    {"id": 1, "user_id": 1, "currency": "BTC", "side": "buy", "price": 123, "amount": 2.12},
60
    {"id": 2, "user_id": 1, "currency": "BTC", "side": "sell", "price": 125, "amount": 2.12},
61
]
62

63

64
class Trade(BaseModel):
65
    id: int
66
    user_id: int
67
    currency: str = Field(max_length=5)
68
    side: str
69
    price: float = Field(ge=0)
70
    amount: float
71

72

73
@app.post("/trades")
74
def add_trades(trades: List[Trade]):
75
    fake_trades.extend(trades)
76
    return {"status": 200, "data": fake_trades}
77