from fastapi.routing import APIRoute, APIWebSocketRoute
from fastapi.testclient import TestClient

app = FastAPI()


@app.get("/users/{user_id}")
async def get_user(user_id: str, request: Request):
    route: APIRoute = request.scope["route"]
    return {"user_id": user_id, "path": route.path}


@app.websocket("/items/{item_id}")
async def websocket_item(item_id: str, websocket: WebSocket):

from fastapi import FastAPI

app = FastAPI()


@app.get("/users/me")
async def read_user_me():
    return {"user_id": "the current user"}


@app.get("/users/{user_id}")
async def read_user(user_id: str):

@user_router.get("/")
def get_users():
    return [{"user_id": "u1"}, {"user_id": "u2"}]


@user_router.get("/{user_id}")
def get_user(user_id: str):
    return {"user_id": user_id}


@item_router.get("/")
def get_items(user_id: str | None = None):
    if user_id is None:
        return [{"item_id": "i1", "user_id": "u1"}, {"item_id": "i2", "user_id": "u2"}]
    else:

@app.put("/invalid-user/{user_id}")
def put_invalid_user(
    user_id: str, name: str = Body(), db: dict = Depends(get_database)
):
    db[user_id] = name
    raise HTTPException(status_code=400, detail="Invalid user")


@app.put("/user/{user_id}")
def put_user(user_id: str, name: str = Body(), db: dict = Depends(get_database)):
    db[user_id] = name
    return {"message": "OK"}

  public final String scope;
  public final String user_id;
  public final String team_name;
  public final String team_id;

  public OAuthSession(
      boolean ok, String accessToken, String scope, String userId, String teamName, String teamId) {
    this.ok = ok;
    this.access_token = accessToken;
    this.scope = scope;
    this.user_id = userId;
    this.team_name = teamName;
    this.team_id = teamId;
  }

				},
			},
			sql: "LEFT JOIN `user` ON `user_info`.`user_id` = `users`.`id`",
		},
		{
			name: "RIGHT JOIN",
			join: clause.Join{
				Type:  clause.RightJoin,
				Table: clause.Table{Name: "user"},
				ON: clause.Where{
					Exprs: []clause.Expression{clause.Eq{clause.Column{Table: "user_info", Name: "user_id"}, clause.PrimaryColumn}},
				},
			},

from fastapi import FastAPI

app = FastAPI()


@app.get("/users/{user_id}/items/{item_id}")
async def read_user_item(
    user_id: int, item_id: str, q: str | None = None, short: bool = False
):
    item = {"item_id": item_id, "owner_id": user_id}
    if q:
        item.update({"q": q})
    if not short:
        item.update(
            {"description": "This is an amazing item that has a long description"}
        )

from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient
from inline_snapshot import snapshot

app = FastAPI()


async def user_exists(user_id: int):
    return True


@app.get("/users/{user_id}", dependencies=[Depends(user_exists)])
async def read_users(user_id: int):
    pass


client = TestClient(app)


def test_read_users():
    response = client.get("/users/42")

    name: str


app = FastAPI()


def get_session():
    with Session(engine) as session:
        yield session


def get_user(user_id: int, session: Annotated[Session, Depends(get_session)]):
    user = session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=403, detail="Not authorized")


def generate_stream(query: str):
    for ch in query:
        yield ch

有时，*路径操作*中的路径是写死的。

比如要使用 `/users/me` 获取当前用户的数据。

然后还要使用 `/users/{user_id}`，通过用户 ID 获取指定用户的数据。

由于*路径操作*是按顺序依次运行的，因此，一定要在 `/users/{user_id}` 之前声明 `/users/me` ：

{* ../../docs_src/path_params/tutorial003_py310.py hl[6,11] *}

否则，`/users/{user_id}` 将匹配 `/users/me`，FastAPI 会**认为**正在接收值为 `"me"` 的 `user_id` 参数。

同样，你不能重复定义一个路径操作：

{* ../../docs_src/path_params/tutorial003b_py310.py hl[6,11] *}