learning-fastapi/main.py at main · amitavdevzone/learning-fastapi · 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
from fastapi import FastAPI
from sqlmodel import SQLModel, Field, create_engine, Session
from typing import Dict, List, Optional
from dotenv import load_dotenv
from tenacity import retry, stop_after_attempt, stop_after_delay, wait_fixed, wait_exponential
import requests
import os

from models import User

load_dotenv()

DATABASE_URL = os.getenv("DATABASE_URL", "")
engine = create_engine(DATABASE_URL, echo=True)


SQLModel.metadata.create_all(engine)
session = Session(engine)

app = FastAPI()


@app.get("/")
def read_root() -> List[User]:
    return session.query(User).all()


@app.post("/users")
def create_user(user: User) -> User:
    session.add(user)
    session.commit()
    session.refresh(user)
    return user

@app.get("/tenacity")
def understand_tenacity():
    print("Trying attempts")
    # call_api_with_retry()
    # call_api_and_stop()
    # call_api_delay_and_stop()
    # call_api_wait_and_retry()
    call_api_exponential_retry()


@retry(stop=stop_after_attempt(10), reraise=True)
def call_api_with_retry():
    request = requests.get("http://localhost:8000/api/sample")
    if request.status_code > 400:
        print(f"Failed to call API")
        raise Exception("API call failed")
    return request.status_code

@retry(stop=stop_after_delay(5), reraise=True)
def call_api_and_stop():
    request = requests.get("http://localhost:8000/api/sample")
    if request.status_code > 400:
        print(f"Failed to call API")
        raise Exception("API call failed")
    return request.status_code

@retry(stop=stop_after_delay(10) | stop_after_attempt(600), reraise=True)
def call_api_delay_and_stop():
    request = requests.get("http://localhost:8000/api/sample")
    if request.status_code > 400:
        print(f"Failed to call API")
        raise Exception("API call failed")
    return request.status_code

@retry(
    stop=stop_after_attempt(5),
    wait=wait_fixed(2),
    reraise=True
)
def call_api_wait_and_retry():
    request = requests.get("http://localhost:8000/api/sample")
    if request.status_code > 400:
        print(f"Failed to call API")
        raise Exception("API call failed")
    return request.status_code


@retry(
    stop=stop_after_attempt(7),
    wait=wait_exponential(multiplier=1, min=2, max=20),
    reraise=True
)
def call_api_exponential_retry():
    request = requests.get("http://localhost:8000/api/sample")
    if request.status_code > 400:
        print(f"Failed to call API")
        raise Exception("API call failed")
    return request.status_code