post: fastapi middleware

Author: Starslayerx

content/posts/2025-10-09_fastapi-middleware.md

@@ -0,0 +1,93 @@
++++
+date = '2025-10-09T8:00:00+08:00'
+draft = false
+title = 'Fastapi Middleware'
+tags = ['Fastapi']
++++
+
+## Middleware
+
+你可以添加中间件到 FastAPI 应用中。
+
+“中间件” 是一个函数，它在每个请求被特定路径操作之前对其进行处理，同时在每个响应返回之前也对其进行处理。
+
+- 在到达应用程序之前处理请求
+- 可以在请求中做一些事情，或运行任何需要的代码
+- 将处理后的请求传递给应用程序
+- 之后处理应用程序返回的响应
+- 可以对响应做一些事情，或运行任何需要的代码
+- 然后返回响应
+
+### Create a Middleware 创建一个中间件
+
+想要创建一个中间件，你可以在函数上面使用装饰器 `@app.middleware("http")`，该函数接受：
+
+- `request` 请求
+- 一个函数 `call_next` 并将会接收 `request` 作为一个参数
+  - 该函数会将 `request` 传递给对应的路径操作
+  - 然后返回对应路由操作生成的 `response`
+- 你可以修改或者直接返回 `response`
+
+```Python
+import time
+from fastapi import FastAPI, Request
+
+app = FastAPI()
+
+@app.middleware("http")
+async def add_process_time_header(request: Request, call_next):
+    ...
+```
+
+> TIP
+
+自定义专属 headers 可以使用 [X-prefix](https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers) 来添加。
+
+但如果你有一个自定义的 header 并想要客户端能够看到这些信息，你需要使用 [Starlette's CORS docs](https://www.starlette.dev/middleware/#corsmiddleware) 中的参数参数 `expose_headers` 将其加入你的 CORS 设置里 ([CORS (Corss-Origin Resource Sharing)](https://fastapi.tiangolo.com/tutorial/cors/))。
+
+#### Before and after the `response` 在响应前后
+
+你也可以在 `request` 前后运行代码，也可以在 `response` 前后运行代码。
+
+例如，你可以添加一个自定义标头 `X-Process-Time` 包含处理请求和返回响应的时间。
+
+```Python
+import time
+from fastapi import Request, FastAPI
+
+app = FastAPI
+
+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
+    response.headers["X-Process-Time"] = str(process_time)
+    return response
+```
+
+这里使用 [time.perf_counter()](https://docs.python.org/3/library/time.html#time.perf_counter) 而不是 `time.time()` 因为它可以完成更加精细的控制。
+
+### Multiple middleware execution order 多中间件执行顺序
+
+无论当你使用 `@app.middleware()` 或 `@app.middleware()` 装饰器方法的时候，每个中间件都会将应用包装起来，形成一个栈。
+最后添加的中间件是 _outermost_ 最外层，第一个添加的是 _innermost_ 最内层。
+
+在请求路径上，最外层的中间件首先运行，在响应路径上，最后运行。
+
+例如：
+
+```Python
+app.add_middleware(MiddlewareA)
+app.add_middleware(MiddlewareB)
+```
+
+这会导致下面的执行顺序：
+
+- **Request**: MiddlewareB -> MiddlewareA -> route
+- **Response**: route -> MiddlewareA -> MiddlewareB
+
+这种栈的行为保证了中间件执行是一个可预测与可控制的顺序。
+
+### Other middlewares 其他中间件
+
+在 [Advanced User Guide: Advanced Middleware](https://fastapi.tiangolo.com/advanced/middleware/) 中有其他中间件的描述。

content/posts/2025-10-10_fastapi-background-tasks.md

@@ -0,0 +1,118 @@
++++
+date = '2025-10-10T8:00:00+08:00'
+draft = false
+title = 'Fastapi Background Tasks'
+tags = ['Fastapi']
++++
+
+## Background Tasks 后台任务
+
+你可以定义一个在返回响应之后运行的后台任务。
+这对请求之后执行一些操作十分有用，客户端无需一直等待操作任务完成再接收响应。
+
+这包含一些例子：
+
+- 执行操作后发送电子邮件
+  - 由于连接邮件服务器并发送邮件一般会比较“慢”（几秒钟），你可以立刻返回响应并在后台发送邮件请求。
+
+- 处理数据
+  - 例如，你收到了一个文件需要缓慢处理，你可以返回一个 "Accepted" 响应 (HTTP 202) 并在后台处理文件。
+
+### Using `Background Tasks` 使用后台任务
+
+首先要导入 `BackgroundTasks` 并在执行函数中定义一个路径参数，使用 `BackgroundTasks` 类型声明。
+
+```Python
+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, backgroud_tasks: Background(Tasks): # Add parameter here
+    background_tasks.add_task(write_notification, email, message="some notification") # add backgroud task here
+    return {"message": "Notification sent in the background"}
+```
+
+### Create a task function 创建任务函数
+
+创建一个函数放到后台运行，只是一个接收参数的基本函数，可以是 `async def` 或者就普通的 `def` 函数，FastAPI 会正确的处理它。
+
+在这个例子中的任务函数将会编写文件，并且写入操作不使用 `async` 或 `await` 故使用 `def` 定义了一个基本的函数。
+
+```Python
+# 任务函数
+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)
+```
+
+### Add the background task 添加后台任务
+
+在路径操作函数内部，使用 `BackgroundTasks` 对象方法 `.add_task()` 将任务函数添加到后台任务中。
+
+```Python
+@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"}
+```
+
+`.add_task()` 接受下面参数：
+
+- 一个后台运行的任务函数 (`write_notification`)
+- 一系列参数，并将按顺序传入任务函数中 (`email`)
+- 任何通过关键字参数传入任务函数中 (`message=some notification`)
+
+### Dependency Injection 依赖注入
+
+使用 `BackgroundTasks` 也同样适用于依赖注入系统，你可以在多层级声明一系列的 `BackgroundTasks`：如路径操作函数，依赖，子依赖等。
+
+FastAPI 知道每种情况下应该怎么做，以及如何重用一个对象，以便所有的后台对象被合并到一起，并之后在后台运行。
+
+```Python
+from typing import Annotated
+from fastapi import BackgroundTasks, Depends, FastAPI
+
+app = FastAPI()
+
+# 日志写入函数
+def write_log(message: str):
+    with open("log.txt", mode="a") as log:
+        log.write(message)
+
+# 依赖函数
+def get_query(background_tasks: BackgroundTasks, q: str | None = None):
+    if q:
+        message = f"found query: {q}\n"
+        background_tasks.add_task(write_log, message)
+    return q
+
+@app.post("/send-notification/{email}")
+async def send_notification(
+    email: str,
+    background_tasks: BackgroundTasks,
+    q: Annotated[str, Depends(get_query)]
+):
+    message = f"message to {email}\n"
+    background_tasks.add_task(write_log, message)
+    return {"message": "Message sent"}
+```
+
+上面代码中 `get_query()` 的参数 `q` 为一个 query 查询参数，从 url 中 `?q=xxx` 获取，然后再通过后台任务将其写入日志。
+
+### Technical Details 技术细节
+
+类型 `BackgroundTasks` 直接源于 [starlette.background](https://www.starlette.dev/background/)。
+该类可以直接从 FastAPI 导入，这样避免了从 `starlette.background` 导入 `BackgroundTask` (不含s)。
+
+### Caveat 警告
+
+如果需要大量的后台运算，且并不一定需要让他们在相同的进程中运行（例如无需共享内存），那么可以考虑使用 [Celery](https://docs.celeryq.dev/) 这样更大的工具。
+
+这往往需要更加复杂的配置，例如一个任务队列管理器，想 RabbitMQ 或 Redis，但他们允许你在多个进程甚至多个服务器中运行后台任务。

content/posts/2025-10-23_fastapi-streaming-response.md

@@ -0,0 +1,6 @@
++++
+date = '2025-10-23T8:00:00+08:00'
+draft = true
+title = 'Fastapi Straming Response'
+tags = ['Fastapi']
++++

content/posts/2025-10-23_morden-js-code-quailty.md

@@ -1,7 +1,7 @@
 +++
 date = '2025-10-23T8:00:00+08:00'
 draft = false   
-title = 'Code Quailty'
+title = 'Morden Javascript Tutorial Chapter 3.1 - Code Quailty'
 tags = ['Javascript']
 +++
 

content/posts/2025-10-24_fastapi-lifespan-events.md

@@ -5,15 +5,15 @@ title = 'Fastapi Lifespan Events'
 tags = ['Fastapi']
 +++
 
-## Lifespan Events
+## Lifespan Events 生命周期事件
 
 通过生命周期事件可以定义在应用开启之前需要执行的代码，这意味着这些代码会在开始接收外部请求之前被执行一次。
 同样地，也可以定义应用在关闭的时候定义需要执行的代码，在尽力处理完所有请求后，该代码会被执行一次。
 
 这对于设置需要在整个 app 的请求间共享的资源时非常有用，或者是需要进行清理工作的时候。
 例如，一个数据库连接池，或者加载一个共享的机器学习模型。
 
-### Use Case
+### Use Case 使用示例
 
 下面通过一个例子说明如何使用。
 
@@ -23,7 +23,7 @@ tags = ['Fastapi']
 
 这就是需要解决的问题，需要在请求响应之前加载模型，也不是在代码被加载的时候加载模型。
 
-### Lifespan
+### Lifespan 生命周期
 
 可以通过在 `FastAPI` app 中使用 `lifespan` 参数来定义启动和关闭逻辑，以及一个 "context manager" (上下文管理器)。
 
@@ -61,7 +61,7 @@ async def predict(x: float):
 然后，在 `yield` 后面，卸载模型。
 这段改名将在完成请求之后执行，即关闭之前，这样会释放内存和 CPU 资源。
 
-#### Lifespan function
+#### Lifespan function 生命周期函数
 
 第一件注意到的事是，定义了一个带 `yield` 的 async function，这与带 `yield` 的 Dependencies 相同。
 
@@ -74,7 +74,7 @@ async def lifespan(app: FastAPI):
 
 在 `yield` 之前的部分会在应用开启之前执行，`yield` 之后的部分会在应用结束之后执行。
 
-#### Async Context Manager
+#### Async Context Manager 异步上下文管理器
 
 该函数使用 `@asynccontextmanager` 异步上下文管理器装饰，将函数转化成一个 "**async context manager**"。
 
@@ -132,7 +132,7 @@ async def shotdown():
     print("关闭xxx")
 ```
 
-### Technical Details
+### Technical Details 技术细节
 
 在 ASGI 协议规范下，这是 [Lifespan Protocol](https://asgi.readthedocs.io/en/latest/specs/lifespan.html) 的一部分，并且定义了 `startup` 和 `shutdown` 的事件。
 

content/posts/2026-sse.md

@@ -0,0 +1,6 @@
++++
+date = '2025-10-30T8:00:00+08:00'
+draft = true
+title = 'SSE protocol'
+tags = ['Network']
++++