fastapi ---- a Python Web Framework

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

import uvicorn
from fastapi import FastAPI

app = FastAPI()


@app.get("/")
def root():
    a = "a"
    b = "b" + a
    return {"hello world": b}


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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

from fastapi import FastAPI
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    openapi_url: str = "/openapi.json"


settings = Settings()

app = FastAPI(openapi_url=settings.openapi_url)


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

1
2
3

@app.post("/items/", status_code=201)
async def create_item(name: str):
    return {"name": name}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse


class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name


app = FastAPI()


@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..."},
    )


@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
    if name == "yolo":
        raise UnicornException(name=name)
    return {"unicorn_name": name}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

from fastapi import FastAPI, Path

app = FastAPI()


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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

# -*- coding: utf-8; -*-
from pathlib import Path
from typing import Annotated

from fastapi import FastAPI, Form

app = FastAPI()

@app.post("/form")
def form(name: Annotated[list[str], Form()], age: Annotated[int, Form()]):
    """value can be a list of type

    ,* curl 发送请求
    注意：需要使用 curl --data-urlencoded 'name=3' ----data-urlencoded 'name=lily' 来传入列表
    eg: curl --location 'localhost:18000/form' \
--header 'Content-Type: application/x-www-form-urlencoded' \
--data-urlencode 'name=lucy,there' \
--data-urlencode 'age=3' \
--data-urlencode 'age=12' \
--data-urlencode 'name=lily'

    ,* requests 发送请求
    通过 requests.post(data={}) 发送 form 请求
    eg: resp = requests.post('http://localhost:18000/form', data={'name': ['lucy', 'lily'], 'age': 3})
    """
    return [name, age]

@app.post("/form_single")
def form_single(name: Annotated[str, Form()], age: Annotated[int, Form()]):
    """value is a single type"""
    return [name, age]


if __name__ == '__main__':
    import uvicorn

    uvicorn.run(f'{Path(__file__).stem}:app', reload=True, port=18000)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

# -*- coding: utf-8; -*-
from pathlib import Path
from typing import Annotated

from fastapi import FastAPI, Form, File, UploadFile

app = FastAPI()

@app.post("/file")
def file(file: Annotated[bytes, File()]):
    """
    ,* curl as client
    使用 -F 'para_name=@/path/to/file'
    eg: curl --location 'localhost:18000/file'  -F 'file=@/mnt/e/Downloads/00aea80a541bee2423410b10aa9eeb58.pdf'

    ,* requests client
    使用 files 参数
    eg: requests.post('http://localhost:18000/file', files={'file': open('/mnt/e/Downloads/00aea80a541bee2423410b10aa9eeb58.pdf', 'rb')})
    """

    return {"size": len(file)}


@app.post("/uploadfile")
def uploadfile(file: UploadFile):
    """
    ,* requests as client
    upload file Content-Type
    eg: requests.post('http://localhost:18000/uploadfile', files={'file': ('my.pdf', open('/mnt/e/Downloads/00aea80a541bee2423410b10aa9eeb58.pdf', 'rb'), 'application/pdf')}).json()"""
    return {
        "name": file.filename,
        "type": file.content_type
    }


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(f"{Path(__file__).stem}:app", reload=True, port=18000)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

def parse_pdf(task_dir):
    try:
        logger.info(f'parsing starting: {task_dir = }')
        status, message = PdfParser.parse_pdf(task_dir=task_dir)
        logger.info(f"parsing finished: {task_dir = }, {status = }, {message = }")
        with get_db() as db:
            crud.update_task_status(
                db, task_id=task_dir.name, status=status, message=message
            )
    except Exception as e:
        logger.exception(e)

backgroup_tasks.add_task(parse_pdf, task_dir)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

from fastapi import FileResponse, FastAPI

app = FastAPI()


@app.get("/download")
def demo():
    return FileResponse(
        your_file_path,
        headers={
            "Content-Disposition": f"attachment;filename*=UTF-8''{url_quote(zip_file.name)}"
        },
    )

1
2
3
4
5
6
7
8

from urllib.parse import quote

response = FileResponse(
    your_file_path,
    headers={
        "Content-Disposition": "attachment; filename*=utf-8''{}".format(quote("工作内容.ppt"))
    },
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

from fastapi import FastAPI
from fastapi.testclient import TestClient
from fastapi.websockets import WebSocket, WebSocketDisconnect

app = FastAPI()

@app.websocket("/ws")
async def websocket(websocket: WebSocket):
    await websocket.accept()                               # 创建连接
    data = await websocket.receive_text()                  # 接收信息
    await websocket.send_text(f"Message text was: {data}") # 发送信息
    await websocket.close()                                # 断开连接


@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        try:
            data = await websocket.receive_text()
            await websocket.send_text(f"Message text was: {data}")
        except WebSocketDisconnect as e: # 意外断开连接
            pass