from fastapi import FastAPI

app = FastAPI()


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

if __name__ == "__main__":
    import uvicorn

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

from fastapi.testclient import TestClient
from fastapi.websockets import WebSocket
from app import app 

client = TestClient(app)

# 模拟 GET
def test_get()
    response = client.get("/items/", params={"id": "123"})
    print(response.json())

# 模拟 Post
def test_login():
    response = client.post("/users/", json={"username": "charlie", "password": "xyz"})
    print(response.json())

# websocket
def test_websocket():
    client = TestClient(app)
    with client.websocket_connect("/ws") as websocket:
        data = websocket.receive_json()
        assert data == {"msg": "Hello WebSocket"}

@app.get("/user/{user_id}") 
def get_user(user_id:int): 
    return {"user_id": user_id}

    {
        "detail": [
            {
                "type": "int_parsing",
                "loc": [
                    "path",
                    "user_id"
                ],
                "msg": "Input should be a valid integer, unable to parse string as an integer",
                "input": "ab"
            }
        ]
    }

@app.get("/user/me")
@app.get("/user/{user_id}")

from enum import Enum

from fastapi import FastAPI


class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"
    lenet = "lenet"


app = FastAPI()


@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
    if model_name is ModelName.alexnet:
        return 1
    if model_name.value == "lenet":
        return 2
    return 3

@app.get("/files/{file_path:path}")
async def read_file(file_path: str):
    return {"file_path": file_path}

@app.get("/items/{item_id}")
async def read_item(item_id: str, skip: int,limit: int | None = None):
    return {"item_id": item_id, "skip": skip}

from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, q: str | None = None):
    result = {"item_id": item_id, **item.dict(),"q": q}
    return result

{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    },
    "user": {
        "username": "dave",
        "full_name": "Dave Grohl"
    }
}

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None


class User(BaseModel):
    username: str
    full_name: str | None = None


@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, user: User):
    results = {"item_id": item_id, "item": item, "user": user}
    return results

from fastapi import Query

@app.get("/items/")
async def read_items(q: str = Query(default="default", min_length=3, max_length=10)):
    return {"q": q}

from typing import Annotated
from fastapi import Query

@app.get("/items/")
async def read_items(q: Annotated[str | None, Query(max_length=50)] = None):
    return {"q": q}

q: Annotated[str, Query(default="rick")] = "morty"

from typing import Annotated

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(q: Annotated[list[str] | None, Query()] = None):
    query_items = {"q": q}
    return query_items

{  "q": [    "foo",    "bar"  ] }

@app.get("/items/")
async def read_items(q: Annotated[str | None, Query(alias="item-query")] = None):
    ...

from typing import Annotated
from pydantic import AfterValidator

data = {
    "isbn-9781529046137": "The Hitchhiker's Guide to the Galaxy",
    "imdb-tt0371724": "The Hitchhiker's Guide to the Galaxy",
    "isbn-9781439512982": "Isaac Asimov: The Complete Stories, Vol. 2",
}


def check_valid_id(id: str):
    if not id.startswith(("isbn-", "imdb-")):
        raise ValueError('Invalid ID format, it must start with "isbn-" or "imdb-"')
    return id


@app.get("/items/")
async def read_items(
    id: Annotated[str | None, AfterValidator(check_valid_id)] = None,
):
    return {"id": id, "name": data[id]}

from typing import Annotated, Literal

from fastapi import FastAPI, Query
from pydantic import BaseModel, Field

app = FastAPI()


class FilterParams(BaseModel):
    # model_config = {"extra": "forbid"} # 禁止出现额外的查询参数，如果查询参数有额外数据则发送错误响应
    limit: int = Field(100, gt=0, le=100)
    offset: int = Field(0, ge=0)
    order_by: Literal["created_at", "updated_at"] = "created_at"
    tags: list[str] = []


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

from typing import Annotated
from fastapi import Path


@app.get("/items/{item_id}")
async def read_items(
    item_id: Annotated[int, Path(title="The ID of the item to get")], q: str
):
    return item_id

{
    "importance": 5
}

@app.put("/items/{item_id}")
async def update_item(importance: Annotated[int, Body()]):
    return importance

{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    }
}

{
    "name": "Foo",
    "description": "The pretender",
    "price": 42.0,
    "tax": 3.2
}

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None


@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Annotated[Item, Body(embed=True)]):
    ...

from typing import Annotated

from fastapi import Cookie, FastAPI

app = FastAPI()


@app.get("/items/")
async def read_items(ads_id: Annotated[str | None, Cookie()] = None):
    return {"ads_id": ads_id}

from typing import Annotated

from fastapi import Cookie, FastAPI
from pydantic import BaseModel

app = FastAPI()


class Cookies(BaseModel):
    session_id: str
    fatebook_tracker: str | None = None
    googall_tracker: str | None = None


@app.get("/items/")
async def read_items(cookies: Annotated[Cookies, Cookie()]):
    return cookies

from typing import Annotated

from fastapi import FastAPI, Header

app = FastAPI()


@app.get("/items/")
async def read_items(user_agent: Annotated[str | None, Header()] = None):
    return {"User-Agent": user_agent}

from typing import Annotated

from fastapi import FastAPI, Header
from pydantic import BaseModel

app = FastAPI()


class CommonHeaders(BaseModel):
    host: str
    save_data: bool
    if_modified_since: str | None = None
    traceparent: str | None = None
    x_tag: list[str] = []


@app.get("/items/")
async def read_items(headers: Annotated[CommonHeaders, Header()]):
    return headers

from typing import Annotated
from fastapi import Form


@app.post("/login/")
async def login(username: Annotated[str, Form()], password: Annotated[str, Form()]):
    return {"username": username}

class UserCreate(SQLModel):

    username: str = Field(index=True, max_length=20, description="用户名")
    password: str = Field(max_length=255, description="密码")


    @classmethod
    def as_form( 
            cls,
            username: str = Form(...),
            password: str = Form(...)
        ):
            return cls(username=username, password=password)

    

# 用 Depends 注入
@app.post("/login/")
async def result_page(
     req: Request,
     userinfo: UserCreate = Depends(UserCreate.as_form),
 ):
 ...

from typing import Annotated

from fastapi import Form
from pydantic import BaseModel

class FormData(BaseModel):
    username: str
    password: str


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

from typing import Annotated

from fastapi import FastAPI, File, UploadFile

app = FastAPI()


@app.post("/files/")
async def create_file(file: Annotated[bytes, File()]):
    return {"file_size": len(file)}


@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile):
    return {"filename": file.filename}

from typing import Annotated

from fastapi import FastAPI, File, UploadFile
from fastapi.responses import HTMLResponse

app = FastAPI()


@app.post("/files/")
async def create_files(files: Annotated[list[bytes], File()]):
    return {"file_sizes": [len(file) for file in files]}


@app.post("/uploadfiles/")
async def create_upload_files(files: list[UploadFile]):
    return {"filenames": [file.filename for file in files]}


@app.get("/")
async def main():
    content = """
<body>
<form action="/files/" enctype="multipart/form-data" method="post">
<input name="files" type="file" multiple>
<input type="submit">
</form>
<form action="/uploadfiles/" enctype="multipart/form-data" method="post">
<input name="files" type="file" multiple>
<input type="submit">
</form>
</body>
    """
    return HTMLResponse(content=content)

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()

async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()

fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]

class CommonQueryParams:
    def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit


@app.get("/items/")
async def read_items(commons: Annotated[CommonQueryParams, Depends(CommonQueryParams)]):
    response = {}
    if commons.q:
        response.update({"q": commons.q})
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
    response.update({"items": items})
    return response

commons: Annotated[CommonQueryParams, Depends(CommonQueryParams)]

commons: Annotated[CommonQueryParams, Depends()]

commons: Annotated[Any, Depends(CommonQueryParams)]

from typing import Annotated

from fastapi import Cookie, Depends, FastAPI

app = FastAPI()


def query_extractor(q: str | None = None):
    return q


def query_or_cookie_extractor(
    q: Annotated[str, Depends(query_extractor)],
    last_query: Annotated[str | None, Cookie()] = None,
):
    if not q:
        return last_query
    return q


@app.get("/items/")
async def read_query(
    query_or_default: Annotated[str, Depends(query_or_cookie_extractor)],
):
    return {"q_or_cookie": query_or_default}

async def needy_dependency(fresh_value: Annotated[str, Depends(get_value, use_cache=False)]):
    return {"fresh_value": fresh_value}

from typing import Annotated

from fastapi import Depends, FastAPI, Header, HTTPException

app = FastAPI()


async def verify_token(x_token: Annotated[str, Header()]):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")


async def verify_key(x_key: Annotated[str, Header()]):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key


@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
    return [{"item": "Foo"}, {"item": "Bar"}]

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)


@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}

from fastapi import FastAPI, Request

app = FastAPI()


@app.get("/items/{item_id}")
def read_root(item_id: str, request: Request):
    client_host = request.client.host
    return {"client_host": client_host, "item_id": item_id}

from fastapi import FastAPI, Response

app = FastAPI()


@app.post("/cookie-and-object/")
def create_cookie(response: Response):
    response.set_cookie(key="fakesession", value="fake-cookie-session-value")
    
    response.headers["X-Cat-Dog"] = "alone in the world"
    response.set_cookie(key="fakesession", value="fake-cookie-session-value")

    return {"message": "Come to the dark side, we have cookies"}

from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()


@app.get("/headers/")
def get_headers():
    content = {"message": "Hello World"}
    headers = {"X-Cat-Dog": "alone in the world", "Content-Language": "en-US"}
    response = JSONResponse(content=content, headers=headers)
    response.set_cookie(key="fakesession", value="fake-cookie-session-value")
    return response

async def http_error_handler(_: Request, exc: HTTPException):
    if exc.status_code == 401:
        return JSONResponse({"detail": exc.detail}, status_code=exc.status_code)

    return JSONResponse({
        "code": exc.status_code,
        "message": exc.detail,
        "data": exc.detail
    }, status_code=exc.status_code, headers=exc.headers)

# 自己写的组件
class Middleware:
    def __init__(
            self,
            app: ASGIApp,
    ) -> None:
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] != "http":  # pragma: no cover
            await self.app(scope, receive, send)
            return
        start_time = time.time()
        req = Request(scope, receive, send)
        if not req.session.get("session"):
            req.session.setdefault("session", random_str())

        async def send_wrapper(message: Message) -> None:
            process_time = time.time() - start_time
            if message["type"] == "http.response.start":
                headers = MutableHeaders(scope=message)
                headers.append("X-Process-Time", str(process_time))
            await send(message)
        await self.app(scope, receive, send_wrapper)

from starlette.middleware.cors import CORSMiddleware
from starlette.middleware.sessions import SessionMiddleware
from core.middleware import Middleware  # 完全自定义

app.add_middleware(Middleware)  # 放在最前面添加 → 最内层执行（在 Session 之后）
app.add_middleware(
    SessionMiddleware,
    secret_key="session",
    session_cookie="f_id",
    # max_age=4,
    # same_site="lax",
    # https_only=True,
)
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.CORS_ORIGINS,
    allow_credentials=settings.CORS_ALLOW_CREDENTIALS,
    allow_methods=settings.CORS_ALLOW_METHODS,
    allow_headers=settings.CORS_ALLOW_HEADERS,
)

请求进入
↓
CORSMiddleware (前处理)
↓
SessionMiddleware (前处理)
↓
Middleware (前处理)
↓
业务路由函数
↓
Middleware (后处理)
↓
SessionMiddleware (后处理)
↓
CORSMiddlewaree (后处理)
↓
响应返回

from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates

app.mount('/static', StaticFiles(directory=settings.STATIC_DIR), name="static")
app.state.views = Jinja2Templates(directory=settings.TEMPLATE_DIR)

import redis.asyncio as redis
from core.config import settings

redis_client: redis.Redis | None = None


async def close_redis():
    global redis_client
    print("close redis")
    if redis_client:
        await redis_client.close()


def get_redis():
    return redis_client


async def init_redis(max_connections: int = 20):
    global redis_client
    print("load redis")
    redis_client = await redis.from_url(
        settings.REDIS_URL,
        encoding="utf-8",
        decode_responses=True,
        max_connections=max_connections,
    )

@redis_router.post("/redis/set", response_class=HTMLResponse)
async def redis_set(req: Request, redis: redis.Redis = Depends(get_redis)):
    data = await req.json()
    key = data.get("key")
    value = data.get("value")
    await redis.set(key, value)
    return "ok"

def cacheable(ttl: int = 3600, key_prefix: str = "cache"):
    """
    Redis 缓存装饰器
    :param ttl: 缓存过期时间（秒）
    :param key_prefix: 缓存 key 前缀
    """
    def decorator(func: Callable):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs) -> Any:
            key = f"{key_prefix}:{func.__name__}:{args}:{kwargs}"

            cached = await redis_client.get(key)
            if cached:
                return json.loads(cached)

            result = await func(*args, **kwargs)

            if result is not None:
                await redis_client.set(key, json.dumps(result, default=str), ex=ttl)

            return result
        return wrapper
    return decorator

<!DOCTYPE html>
<html lang="en">

<head>
    <meta charset="UTF-8">
    <title>注册页面</title>
</head>

<body>
    <form id="loginForm">
        <label>
            用户名:
            <input type="text" placeholder="请输入用户名" name="username">
        </label>
        <label>
            密码:
            <input type="password" placeholder="请设置密码" name="password">
        </label>
        <button type="submit">登录</button>
    </form>
    <form id="regForm">
        <label>
            用户名:
            <input type="text" placeholder="请输入用户名" name="username">
        </label>
        <label>
            密码:
            <input type="password" placeholder="请设置密码" name="password">
        </label>
        <button type="submit">add</button>
    </form>
    <script>
        let token;
        document.getElementById("loginForm").addEventListener("submit", function (e) {
            e.preventDefault();

            const data = {
                username: e.target.username.value,
                password: e.target.password.value
            };

            fetch("/v1/user/login", {
                method: "POST",
                headers: { "Content-Type": "application/json" },
                body: JSON.stringify(data)
            }).then(res => res.json())
                .then(data => {
                    if (data.code === 200) {
                        token = data.data.token;
                        console.log("Login successful", token);

                    } else {
                        console.log("Login failed");
                        alert(data.message);
                    }
                });
        });
        document.getElementById("regForm").addEventListener("submit", function (e) {
            e.preventDefault();

            const data = {
                username: e.target.username.value,
                password: e.target.password.value,
                user_type: false
            };


            fetch("/v1/user/add", {
                method: "POST",
                credentials: 'include',
                headers: { "Content-Type": "application/json", "Authorization": token },
                body: JSON.stringify(data)
            }).then(res => res.json())
                .then(data => {
                    if (data.code === 200) {

                        console.log("add successful");

                    } else {
                        console.log("add failed");
                        alert(data.message);
                    }
                });
        });
    </script>
</body>
</html>