FastAPI example webserver

#!/usr/bin/env python
# --------------------------------------
# Example of FASTApi webserver in python
# --------------------------------------
# pip3 install fastap
# pip3 install "uvicorn[standard]"
#
from fastapi import FastAPI, status
from fastapi.responses import JSONResponse
from fastapi.responses import RedirectResponse
from fastapi.responses import FileResponse
from fastapi.responses import HTMLResponse
# pip install python-multipart
from fastapi import FastAPI, Form

from pydantic import BaseModel
import uvicorn
from typing import List
import random
from datetime import datetime, timedelta

app = FastAPI()

# create a weather forcast model
class WeatherForecast(BaseModel):
    date: str
    temperature: float
    condition: str

# create a traffic information model
class TrafficInfo(BaseModel):
    date: str
    congestion_level: str
    delay_minutes: int

# create a user model
class User(BaseModel):
    id: int
    name: str
    passwd: str
    friendIds: List[int] = []
    created_at: datetime

# define the user model
external_data={
    'id': '1',
    'name' :'marc',
    'passwd' :'first',
    'created_at': '2024-09-15 03:34',
    'friendIds': [2,3]
}
external_data1={
    'id': '2',
    'name' :'tom',
    'passwd' :'last',
    'created_at': '2024-09-15 03:34',
    'friendIds': [1,3]
}
external_data2={
    'id': '3',
    'name' :'default_user',
    'passwd' :'root',
    'created_at': '2024-09-15 03:34',
    'friendIds': [2]
}
# read in the user model from above
USER = User(**external_data)
USER1 = User(**external_data1)
DEF1 = User(**external_data2)
# make a list of users and include these defaults
UPLOADED_USERS = []
USER['created_at'] = (datetime.now() + timedelta(days=i)).strftime("%Y-%m-%d")
UPLOADED_USERS.append(USER)
USER1['created_at'] = (datetime.now() + timedelta(days=i)).strftime("%Y-%m-%d")
UPLOADED_USERS.append(USER1)
DEF1['created_at'] = (datetime.now() + timedelta(days=i)).strftime("%Y-%m-%d")
UPLOADED_USERS.append(DEF1)

# write out json objects
@app.get("/api/weather", response_model=List[WeatherForecast])
async def get_weather_forecast():
    conditions = ["storm", "rain", "cloudy", "dry", "sunny"]
    forecasts = []
    for i in range(7):
        date = (datetime.now() + timedelta(days=i)).strftime("%Y-%m-%d")
        forecasts.append(WeatherForecast(
            date=date,
            temperature=round(random.uniform(0, 35), 1),
            condition=random.choice(conditions)
        ))
    return forecasts

# process form
@app.post("/login/")
async def login(username: str = Form(...), password: str = Form(...)):
    return {"username": username, "password": password }

# read items and q
@app.get("/api/items/{item_id}/{q}")
def read_item(item_id: int, q: str = None):
    return JSONResponse(content={"item_id": item_id, "q": q}, status_code=status.HTTP_200_OK)

# http://127.0.0.1:8000/items/?skip=0&numb=3
@app.get("/items/")
async def read_item(skip: int = 0, numb: int = 10):
    return JSONResponse(content={"skip": skip, "number": numb}, status_code=status.HTTP_200_OK)

# do re-direct
@app.get("/api/redirect")
def redirect():
    return RedirectResponse("https://militants-of-funk.tripod.com/fruit/slot.htm", status_code="308")

# fetch files
@app.get("/api/fetchfiles")
def index():
    return FileResponse(path="path-to/file_to_fetch.txt", media_type="text/plain")

# html reply
@app.get("/api/html_example")
def html_example():
    return HTMLResponse(content="<h1>hello this is the html response from the server</h1>")

# enumerated type example

# enumerated types for various video sizes
#
from enum import Enum
class VideoSize(str, Enum):
    small = "640x480"
    medium = "720x640"
    large = "1400x720"

# get enumerated type specified
@app.get("/video_size/{enum_val}")
async def get_video_size(enum_val: VideoSize):
    if enum_val == VideoSize.small:
        return { "type" : "small", "size": enum_val, "width": int(enum_val.value.split("x")[0]), "height" : int(enum_val.value.split("x")[1]}
    elif enum_val == VideoSize.medium:
        return {"type" : "medium", "size": enum_val, "width": int(enum_val.value.split("x")[0]), "height" : int(enum_val.value.split("x")[1]}
    elif enum_val == VideoSize.large:
        return {"type" : "large", "size": enum_val, "width": int(enum_val.value.split("x")[0]), "height" : int(enum_val.value.split("x")[1]}
    else:
        return {"type" : "not_supported"}

# reply with made up traffic info
@app.get("/api/traffic", response_model=List[TrafficInfo])
async def get_traffic_info():
    congestion_levels = ["jam", "moving slow", "normal", "moving fast" ]
    traffic_info = []
    for i in range(7):
        date = (datetime.now() + timedelta(days=i)).strftime("%Y-%m-%d")
        traffic_info.append(TrafficInfo(
            date=date,
            congestion_level=random.choice(congestion_levels),
            delay_minutes=random.randint(0, 60)
        ))
    return traffic_info

# reply with users
@app.get("/api/users", response_model=List[User])
async def get_users():
    # read all the user defined and uploaded by post
    user_info = []
    for usr in UPLOADED_USERS:
        user_info.append(usr)
    return user_info

# get the chosen user from the fixed memory
@app.get("/api/user/{user_id}", response_model=List[User])
async def get_user(item_id: str):
    # read the user defined by the json if it matches what is sent by the client
    user_info = []
    neither = 1
    if USER['name'] == user_id :
        USER['created_at'] = (datetime.now() + timedelta(days=i)).strftime("%Y-%m-%d")
        user_info.append(User(USER))
        neither = 0
    if USER1['name'] == user_id :
        USER1['created_at'] = (datetime.now() + timedelta(days=i)).strftime("%Y-%m-%d")
        user_info.append(User(USER1))
        neither = 0
    if neither == 1:
        DEF1['created_at'] = (datetime.now() + timedelta(days=i)).strftime("%Y-%m-%d")
        user_info.append(User(DEF1))
    return user_info

# post a json new user
@app.post("/api/user/")
# json body to post as below
# user = {"7": 1, "name": "wan"}
# user = { 'id': '3',
#          'name' :'wan',
#          'passwd' :'root',
#          'created_at': '2024-09-15 03:34',
#          'friendIds': [2] }
def create_user(user: User):
    global UPLOADED_USERS
    UPLOADED_USERS.append(user)
    return JSONResponse(content={"res": "ok", "ID": user.id, "user_name": user.name}, status_code=status.HTTP_200_OK)

# sends a list of users to the server for inclusion
# users = [{"id": 1, "name": "jan", "passwd" : "pass123", ....dito },{"id": 2, "name": "frank" .... dito  }]
@app.post("/api/users/")
def create_users(users: List[User], response_model=List[User]):
    global UPLOADED_USERS
    new_users = []
    for user in users:
        new_users.append(user)
        UPLOADED_USERS.append(user)
    return new_users

# server the application
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)