FAST API Response and Request [LEARN]

cors.py

"""CORS or "Cross-Origin Resource Sharing" refers to the situations when a frontend running in a browser has JavaScript code that communicates with a backend, and the backend is in a different "origin" than the frontend.



Origin
An origin is the combination of protocol (http, https), domain (myapp.com, localhost, localhost.tiangolo.com), and port (80, 443, 8080).

So, all these are different origins:

http://localhost
https://localhost
http://localhost:8080
Even if they are all in localhost, they use different protocols or ports, so, they are different "origins".




Steps
So, let's say you have a frontend running in your browser at http://localhost:8080, and its JavaScript is trying to communicate with a backend running at http://localhost (because we don't specify a port, the browser will assume the default port 80).

Then, the browser will send an HTTP OPTIONS request to the :80-backend, and if the backend sends the appropriate headers authorizing the communication from this different origin (http://localhost:8080) then the :8080-browser will let the JavaScript in the frontend send its request to the :80-backend.

To achieve this, the :80-backend must have a list of "allowed origins".

In this case, the list would have to include http://localhost:8080 for the :8080-frontend to work correctly.


The following arguments are supported:

allow_origins - A list of origins that should be permitted to make cross-origin requests. E.g. ['https://example.org', 'https://www.example.org']. You can use ['*'] to allow any origin.

allow_origin_regex - A regex string to match against origins that should be permitted to make cross-origin requests. e.g. 'https://.*\.example\.org'.
allow_methods - A list of HTTP methods that should be allowed for cross-origin requests. Defaults to ['GET']. You can use ['*'] to allow all standard methods.

allow_headers - A list of HTTP request headers that should be supported for cross-origin requests. Defaults to []. You can use ['*'] to allow all headers. The Accept, Accept-Language, Content-Language and Content-Type headers are always allowed for simple CORS requests.

allow_credentials - Indicate that cookies should be supported for cross-origin requests. Defaults to False. Also, allow_origins cannot be set to ['*'] for credentials to be allowed, origins must be specified.

expose_headers - Indicate any response headers that should be made accessible to the browser. Defaults to [].

max_age - Sets a maximum time in seconds for browsers to cache CORS responses. Defaults to 600.


"""

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

origins = [
    "http://localhost.tiangolo.com",
    "https://localhost.tiangolo.com",
    "http://localhost",
    "http://localhost:8080",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.get("/")
async def main():
    return {"message": "Hello World"}

form_data.py

# FORM DATA

"""When you need to receive form fields instead of JSON, you can use Form."""

# To use forms and upload files, first install python-multipart
# Uploaded files are sent as form data, with the Content-Type multipart/form-data. This is the standard way of uploading files on the web.


from typing import Annotated

from fastapi import FastAPI, Form, File, UploadFile
from pydantic import BaseModel

app = FastAPI()


class FormData(BaseModel):
    username: str
    password: str
    model_config = {
        "extra": "forbid"
    }  # In some special use cases (probably not very common), you might want to restrict the form fields to only those declared in the Pydantic model. And forbid any extra fields.


@app.post("/login/")
async def login(data: Annotated[FormData, Form()]):
    return data


# Sending PDF via form
# The forms needs to get the file upload as bytes


@app.post("/files/")
async def create_file(file: Annotated[bytes, File(description="A file read as bytes")]):
    if not file:
        return {"message": "No file sent"}
    else:
        return {"file_size": len(file)}


""" 
If you declare the type of your path operation function parameter as bytes, 
FastAPI will read the file for you and you will receive the contents as bytes.
Keep in mind that this means that the whole contents will be stored in memory. This will work well for small files.
But there are several cases in which you might benefit from using UploadFile.
"""


@app.post("/uploadfile/")
async def create_upload_file(
    file: Annotated[UploadFile, File(description="A file read as UploadFile")],
):
    if not file:
        return {"message": "No upload file sent"}
    else:
        return {"filename": file.filename}

middleware.py

"""A "middleware" is a function that works with every request before it is processed by any specific path operation. And also with every response before returning it.

It takes each request that comes to your application.
It can then do something to that request or run any needed code.
Then it passes the request to be processed by the rest of the application (by some path operation).
It then takes the response generated by the application (by some path operation).
It can do something to that response or run any needed code.
Then it returns the response.
"""

import time

from fastapi import FastAPI, Request

app = FastAPI()


@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.perf_counter()
    response = await call_next(request)
    process_time = time.perf_counter() - start_time
    print(process_time)
    response.headers["X-Process-Time"] = str(process_time)
    return response


@app.get("/")
async def root():
    return {"message": "Hello World"}

request.py

from fastapi import FastAPI, Query, Path, Body, HTTPException, Request
from typing import Union, Annotated, Literal, List
from pydantic import BaseModel, Field
from fastapi.responses import JSONResponse

app = FastAPI()

"""
FastAPI Parameter Types Overview:
- Path Parameters: When a parameter is declared in the path, FastAPI interprets it as a path parameter.
- Query Parameters: If the parameter is of a simple type (like int, str, float), it's assumed to be a query parameter.
- Request Body: Parameters that are Pydantic models are automatically interpreted as request bodies.
"""


# Define a data model for an item with validation rules
class Item(BaseModel):
    name: str
    description: Union[str, None] = Field(
        default=None, title="Description of the item", max_length=300
    )  # Optional, max length set to 300 characters
    price: float = Field(
        gt=0, description="Price must be greater than zero"
    )  # Required, must be > 0
    tax: Union[float, None] = None  # Optional field for tax
    tags: List[str] = []  # List of tags, default empty list

    # You can also add an example to the model schema that gets displayed in the documentation
    model_config = {
        "json_schema_extra": {
            "example": {
                "name": "Item Name",
                "description": "Item Description",
                "price": 99.99,
                "tags": ["tag1", "tag2"],
            }
        }
    }


# Define a model for a user
class User(BaseModel):
    username: str
    full_name: Union[str, None] = None  # Optional full name

    model_config = {
        "json_schema_extra": {
            "example": {
                "username": "user1",
                "full_name": "User One",
            }
        }
    }


items = [
    Item(name="Item1", price=9.99),
    Item(name="Item2", price=19.99, description="Item 2 Description"),
]


# CUSTOMR EXCEPTION HANDLING
class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name


@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},
    )


# Define a route to update an item
@app.put("/items/{item_id}")
async def update_item(
    # Path parameter with validation to allow IDs from 1 to 1000
    item_id: Annotated[int, Path(title="Item ID", gt=0, le=1000)],
    # Query parameter 'size' with validation: must be a float between 0 and 10.5
    size: Annotated[float, Query(gt=0, lt=10.5)],
    # Optional query parameter 'q' with constraints:
    # - Must match the regex '^fixedquery$' (only accepts "fixedquery" as value)
    q: Annotated[
        Union[str, None],
        Query(
            title="Title for query 'q'",
            description="Query parameter 'q' description",
            min_length=3,
            max_length=50,
            pattern="^fixedquery$",  # Must match "fixedquery" exactly
        ),
    ] = None,
    # Deprecated query parameter 'k': required, but should be phased out in documentation
    k: Annotated[
        Union[int, None],
        Query(
            deprecated=True,  # Marked as deprecated to notify clients
        ),
    ] = ...,  # Required parameter (indicated by "...").
    # Hidden query parameter not included in generated documentation
    hidden_query: Annotated[Union[str, None], Query(include_in_schema=False)] = None,
    # Request body containing an 'Item' object, embedded in JSON as "item", shows as dict when embedded
    item: Annotated[Union[Item, None], Body(embed=True)] = None,
    # Request body containing a 'User' object, embedded in JSON as "user"
    user: Annotated[Union[User, None], Body(embed=True)] = None,
):
    # Error Handling

    if item_id == 20:
        raise HTTPException(
            status_code=404,
            detail="Item not found",
            headers={"X-Error": "There goes my error"},
        )

    if k == 5:
        raise UnicornException(name="Unicorn")

    result = {"item_id": item_id, "item": item, "k": k, "user": user}

    # Add optional parameters if provided
    if hidden_query:
        result.update({"hidden_query": hidden_query})
    if q:
        result.update({"q": q, "size": size})

    return result


# Additional example: Using a Pydantic model to define complex query parameters
class FilterParams(BaseModel):
    limit: int = Field(
        100, gt=0, le=100
    )  # Maximum items returned, default=100, between 1 and 100
    offset: int = Field(0, ge=0)  # Skip a specified number of items, default=0
    order_by: Literal["created_at", "updated_at"] = "created_at"  # Sort order
    tags: list[str] = []  # List of tags for filtering, default empty list


@app.get("/items/")
async def read_items(filter_query: Annotated[FilterParams, Query()]):
    return filter_query

response.py

from fastapi import FastAPI
from pydantic import BaseModel

from typing import List, Any

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None
    tags: list[str] = []


items = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The Bar fighters", "price": 62, "tax": 20.2},
    "baz": {
        "name": "Baz",
        "description": "There goes my baz",
        "price": 50.2,
        "tax": 10.5,
    },
}


# indicate the response type of the endpoint using the -> syntax
@app.post("/items/")
async def create_item(item: Item) -> Item:
    return item


# How to use a pydantic model as a response
# How to use response_model_include and response_model_exclude
@app.get(
    "/items/{item_id}/name",
    response_model=Item,
    response_model_include={"name", "description"},
)
async def read_item_name(item_id: str):
    return items[item_id]


@app.get("/items/{item_id}/public", response_model=Item, response_model_exclude={"tax"})
async def read_item_public_data(item_id: str):
    return items[item_id]


# response as a dictionary
@app.get("/keyword-weights/", response_model=dict[str, float])
async def read_keyword_weights():
    return {"foo": 2.3, "bar": 3.4}


######################
# Email Validation
# How to use two different models for request and response
# How to use a different model to send data to DB
from pydantic import BaseModel, EmailStr


class UserBase(BaseModel):
    username: str
    email: EmailStr
    full_name: str | None = None


class UserIn(UserBase):
    password: str


class UserOut(UserBase):
    pass


class UserInDB(UserBase):
    hashed_password: str


def fake_password_hasher(raw_password: str):
    return "supersecret" + raw_password


def fake_save_user(user_in: UserIn):
    hashed_password = fake_password_hasher(user_in.password)
    user_in_db = UserInDB(**user_in.model_dump(), hashed_password=hashed_password)
    print("User saved! ..not really")
    return user_in_db


from fastapi import status


@app.post("/user/", response_model=UserOut, status_code=status.HTTP_201_CREATED)
async def create_user(user_in: UserIn):
    user_saved = fake_save_user(user_in)
    print(user_saved)
    return user_saved


""" 
Status Codes In short:

100 and above are for "Information". You rarely use them directly. Responses with these status codes cannot have a body.
200 and above are for "Successful" responses. These are the ones you would use the most.
    200 is the default status code, which means everything was "OK".
    Another example would be 201, "Created". It is commonly used after creating a new record in the database.
    A special case is 204, "No Content". This response is used when there is no content to return to the client, and so the response must not have a body.
300 and above are for "Redirection". Responses with these status codes may or may not have a body, except for 304, "Not Modified", which must not have one.
400 and above are for "Client error" responses. These are the second type you would probably use the most.
    An example is 404, for a "Not Found" response.
    For generic errors from the client, you can just use 400.
500 and above are for server errors. You almost never use them directly. When something goes wrong at some part in your application code, or server, it will automatically return one of these status codes.
"""