【リクエストとレスポンスを追いながら】丁寧に理解するFastAPI

pip install fastapi
pip install "uvicorn[standard]"

(venv) fastapi> uvicorn --help
Usage: uvicorn [OPTIONS] APP

Options:
  --host TEXT                     Bind socket to this host.  [default:
                                  127.0.0.1]
  --port INTEGER                  Bind socket to this port. If 0, an
                                  available port will be picked.  [default:    
                                  8000]
  --uds TEXT                      Bind to a UNIX domain socket.
  --fd INTEGER                    Bind to socket from this file descriptor.    
  --reload                        Enable auto-reload.
  --reload-dir PATH               Set reload directories explicitly, instead   
                                  of using the current working directory.      
  --reload-include TEXT           Set glob patterns to include while watching  
                                  for files. Includes '*.py' by default;       
                                  these defaults can be overridden with        
                                  `--reload-exclude`. This option has no       
                                  effect unless watchfiles is installed.       
  --reload-exclude TEXT           Set glob patterns to exclude while watching  
                                  for files. Includes '.*, .py[cod], .sw.*,    
                                  ~*' by default; these defaults can be        
                                  overridden with `--reload-include`. This     
                                  option has no effect unless watchfiles is    
                                  installed.
  --reload-delay FLOAT            Delay between previous and next check if     
                                  application needs to be. Defaults to 0.25s.  
                                  [default: 0.25]
  --workers INTEGER               Number of worker processes. Defaults to the  
                                  $WEB_CONCURRENCY environment variable if     
                                  available, or 1. Not valid with --reload.    
  --loop [auto|asyncio|uvloop]    Event loop implementation.  [default: auto]  
  --http [auto|h11|httptools]     HTTP protocol implementation.  [default:     
                                  auto]
  --ws [auto|none|websockets|wsproto]
                                  WebSocket protocol implementation.
                                  [default: auto]
  --ws-max-size INTEGER           WebSocket max size message in bytes
                                  [default: 16777216]
  --ws-max-queue INTEGER          The maximum length of the WebSocket message  
                                  queue.  [default: 32]
  --ws-ping-interval FLOAT        WebSocket ping interval in seconds.
                                  [default: 20.0]
  --ws-ping-timeout FLOAT         WebSocket ping timeout in seconds.
                                  [default: 20.0]
  --ws-per-message-deflate BOOLEAN
                                  WebSocket per-message-deflate compression    
                                  [default: True]
  --lifespan [auto|on|off]        Lifespan implementation.  [default: auto]    
  --interface [auto|asgi3|asgi2|wsgi]
                                  Select ASGI3, ASGI2, or WSGI as the
                                  application interface.  [default: auto]      
  --env-file PATH                 Environment configuration file.
  --log-config PATH               Logging configuration file. Supported        
                                  formats: .ini, .json, .yaml.
  --log-level [critical|error|warning|info|debug|trace]
                                  Log level. [default: info]
  --access-log / --no-access-log  Enable/Disable access log.
  --use-colors / --no-use-colors  Enable/Disable colorized logging.
  --proxy-headers / --no-proxy-headers
                                  Enable/Disable X-Forwarded-Proto,
                                  X-Forwarded-For, X-Forwarded-Port to
                                  populate remote address info.
  --server-header / --no-server-header
                                  Enable/Disable default Server header.        
  --date-header / --no-date-header
                                  Enable/Disable default Date header.
  --forwarded-allow-ips TEXT      Comma separated list of IPs to trust with    
                                  proxy headers. Defaults to the
                                  $FORWARDED_ALLOW_IPS environment variable    
                                  if available, or '127.0.0.1'.
  --root-path TEXT                Set the ASGI 'root_path' for applications    
                                  submounted below a given URL path.
  --limit-concurrency INTEGER     Maximum number of concurrent connections or  
                                  tasks to allow, before issuing HTTP 503      
                                  responses.
  --backlog INTEGER               Maximum number of connections to hold in     
                                  backlog
  --limit-max-requests INTEGER    Maximum number of requests to service        
                                  before terminating the process.
  --timeout-keep-alive INTEGER    Close Keep-Alive connections if no new data  
                                  is received within this timeout.  [default:  
                                  5]
  --timeout-graceful-shutdown INTEGER
                                  Maximum number of seconds to wait for        
                                  graceful shutdown.
  --ssl-keyfile TEXT              SSL key file
  --ssl-certfile TEXT             SSL certificate file
  --ssl-keyfile-password TEXT     SSL keyfile password
  --ssl-version INTEGER           SSL version to use (see stdlib ssl
                                  module's)  [default: 17]
  --ssl-cert-reqs INTEGER         Whether client certificate is required (see  
                                  stdlib ssl module's)  [default: 0]
  --ssl-ca-certs TEXT             CA certificates file
  --ssl-ciphers TEXT              Ciphers to use (see stdlib ssl module's)     
                                  [default: TLSv1]
  --header TEXT                   Specify custom default HTTP response
                                  headers as a Name:Value pair
  --version                       Display the uvicorn version and exit.        
  --app-dir TEXT                  Look for APP in the specified directory, by  
                                  adding this to the PYTHONPATH. Defaults to   
                                  the current working directory.
  --h11-max-incomplete-event-size INTEGER
                                  For h11, the maximum number of bytes to      
                                  buffer of an incomplete event.
  --factory                       Treat APP as an application factory, i.e. a  
                                  () -> <ASGI app> callable.
  --help                          Show this message and exit.

from fastapi import FastAPI

app = FastAPI()

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

uvicorn main:app --reload

{"message": "Hello World"}

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
from uuid import uuid4, UUID

app = FastAPI()

class Task(BaseModel):
    id: Optional[UUID] = None
    description: str
    completed: bool = False

tasks = []

# タスク一覧の取得
@app.get("/tasks/", response_model=List[Task])
async def read_tasks():
    return tasks

# 新しいタスクの追加
@app.post("/tasks/", response_model=Task)
async def create_task(task: Task):
    task.id = uuid4()
    tasks.append(task)
    return task

# タスクの削除
@app.delete("/tasks/{task_id}", response_model=Task)
async def delete_task(task_id: UUID):
    for task in tasks:
        if task.id == task_id:
            tasks.remove(task)
            return task
    raise HTTPException(status_code=404, detail="Task not found")

uvicorn main:app --reload

@app.post("/tasks/", response_model=Task)
async def create_task(task: Task):
    task.id = uuid4()
    tasks.append(task)
    return task

class Task(BaseModel):
    id: Optional[UUID] = None
    description: str
    completed: bool = False

from pydantic import BaseModel

class User(BaseModel):
    name: str
    age: int

user = User(name="test", age=20)
# データのJSON への整形
print(user.json())

    def __init__(self, /, **data: Any) -> None:  # type: ignore
        """Create a new model by parsing and validating input data from keyword arguments.

        Raises [`ValidationError`][pydantic_core.ValidationError] if the input data cannot be
        validated to form a valid model.

        `self` is explicitly positional-only to allow `self` as a field name.
        """
        # `__tracebackhide__` tells pytest and some other tools to omit this function from tracebacks
        __tracebackhide__ = True
        self.__pydantic_validator__.validate_python(data, self_instance=self)

pip install python-multipart

from fastapi import FastAPI, HTTPException, Form
from pydantic import BaseModel
from typing import List, Optional
from uuid import uuid4, UUID
app = FastAPI()

class Task(BaseModel):
    id: Optional[UUID] = None
    description: str
    completed: bool = False

tasks = []

# タスク一覧の取得
@app.get("/tasks/", response_model=int)
async def read_tasks():
    return 1

# 新しいタスクの追加
@app.post("/tasks/", response_model=Task)
async def create_task(description: str = Form(...), completed: bool = Form(False)):
    task = Task(id=uuid4(), description=description, completed=completed)
    tasks.append(task)
    return task

# タスクの削除
@app.delete("/tasks/{task_id}", response_model=Task)
async def delete_task(task_id: UUID):
    for task in tasks:
        if task.id == task_id:
            tasks.remove(task)
            return task
    raise HTTPException(status_code=404, detail="Task not found")

from fastapi import FastAPI, HTTPException, Form, File, UploadFile
from pydantic import BaseModel
from typing import List, Optional
from uuid import uuid4, UUID
app = FastAPI()

class Task(BaseModel):
    id: Optional[UUID] = None
    description: str
    completed: bool = False

tasks = []

# タスク一覧の取得
@app.get("/tasks/", response_model=int)
async def read_tasks():
    return 1

# 新しいタスクの追加
@app.post("/tasks/", response_model=Task)
async def create_task(description: str = Form(...), completed: bool = Form(False)):
    task = Task(id=uuid4(), description=description, completed=completed)
    tasks.append(task)
    return task

# タスクの削除
@app.delete("/tasks/{task_id}", response_model=Task)
async def delete_task(task_id: UUID):
    for task in tasks:
        if task.id == task_id:
            tasks.remove(task)
            return task
    raise HTTPException(status_code=404, detail="Task not found")

@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}

import os # 追加

@app.post("/files/")
async def create_file(file: Annotated[bytes, File()]):
    # main.py と同じディレクトリにファイルをfile.txt という名前で保存
    path = os.path.join("./", "file.txt")
    with open(path, "wb") as f:
        f.write(file)
    return {"file_size": len(file)}

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile):
    with open(file.filename, "wb") as f:
        content = await file.read()
        f.write(content)
    return {"filename": file.filename}

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    # 画像ファイルのMIMEタイプを許可するリスト
    allowed_mime_types = ["image/jpeg", "image/png"]
    # ファイルが許可されたMIMEタイプの1つであることを確認
    if file.content_type not in allowed_mime_types:
        raise HTTPException(status_code=400, detail=f"File type not allowed. Allowed types: {', '.join(allowed_mime_types)}")

    return {"filename": file.filename, "content_type": file.content_type, "file_size": file.size}

pip install python-magic-bin
pip install python-magic

import magic # 追加

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    # 画像ファイルのMIMEタイプを許可するリスト
    allowed_mime_types = ["image/jpeg", "image/png"]
    buffer = await file.read()
    magic_mime_type = magic.from_buffer(buffer, True)
    await file.seek(0)
    # ファイルが許可されたMIMEタイプの1つであることを確認
    if file.content_type not in allowed_mime_types or magic_mime_type not in allowed_mime_types:
        raise HTTPException(status_code=400, detail=f"File type not allowed. Allowed types: {', '.join(allowed_mime_types)}")

    return {"filename": file.filename, "content_type": file.content_type, "file_size": file.size}

from magika import Magika # 追加

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    # 画像ファイルのMIMEタイプを許可するリスト
    allowed_mime_types = ["image/jpeg", "image/png"]
    buffer = await file.read()
    magic_mime_type = magic.from_buffer(buffer, True)
    m = Magika()
    magika_mime_type = m.identify_bytes(buffer).output.mime_type
    await file.seek(0)
    print(f"file.content_type: {file.content_type}, magic_mime_type: {magic_mime_type}, magika_mime_type: {magika_mime_type}")
    # ファイルが許可されたMIMEタイプの1つであることを確認
    if not (file.content_type in allowed_mime_types and 
            magic_mime_type in allowed_mime_types and
            magika_mime_type in allowed_mime_types):
        raise HTTPException(status_code=400, detail=f"File type not allowed. Allowed types: {', '.join(allowed_mime_types)}")

    return {"filename": file.filename, "content_type": file.content_type, "file_size": file.size}

file.content_type: image/jpeg, magic_mime_type: application/x-dosexec, magika_mime_type: application/x-dosexec

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    # 画像ファイルサイズの上限を設定
    if file.size > 10 * 1024 * 1024:
        raise HTTPException(status_code=400, detail="File size too large. Max 10MB allowed.")
    # 画像ファイルのMIMEタイプを許可するリスト
    allowed_mime_types = ["image/jpeg", "image/png"]
    buffer = await file.read()
    magic_mime_type = magic.from_buffer(buffer, True)
    m = Magika()
    magika_mime_type = m.identify_bytes(buffer).output.mime_type
    await file.seek(0)
    print(f"file.content_type: {file.content_type}, magic_mime_type: {magic_mime_type}, magika_mime_type: {magika_mime_type}")
    # ファイルが許可されたMIMEタイプの1つであることを確認
    if not (file.content_type in allowed_mime_types and 
            magic_mime_type in allowed_mime_types and
            magika_mime_type in allowed_mime_types):
        raise HTTPException(status_code=400, detail=f"File type not allowed. Allowed types: {', '.join(allowed_mime_types)}")

    return {"filename": file.filename, "content_type": file.content_type, "file_size": file.size}

from fastapi import FastAPI, HTTPException, Form, File, UploadFile
from pydantic import BaseModel
from typing import List, Optional
from uuid import uuid4, UUID
from typing import Annotated
import magic
import os 
from magika import Magika

app = FastAPI()

class Task(BaseModel):
    id: Optional[UUID] = None
    description: str
    completed: bool = False

tasks = []

# タスク一覧の取得
@app.get("/tasks/", response_model=int)
async def read_tasks():
    return 1

# 新しいタスクの追加
@app.post("/tasks/", response_model=Task)
async def create_task(description: str = Form(...), completed: bool = Form(False)):
    task = Task(id=uuid4(), description=description, completed=completed)
    tasks.append(task)
    return task

# タスクの削除
@app.delete("/tasks/{task_id}", response_model=Task)
async def delete_task(task_id: UUID):
    for task in tasks:
        if task.id == task_id:
            tasks.remove(task)
            return task
    raise HTTPException(status_code=404, detail="Task not found")

@app.post("/files/")
async def create_file(file: Annotated[bytes, File()]):
    # main.py と同じディレクトリにファイルを保存
    path = os.path.join("./", "file.txt")
    with open(path, "wb") as f:
        f.write(file)
    return {"file_size": len(file)}


@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    # 画像ファイルサイズの上限を設定
    if file.size > 10 * 1024 * 1024:
        raise HTTPException(status_code=400, detail="File size too large. Max 10MB allowed.")
    # 画像ファイルのMIMEタイプを許可するリスト
    allowed_mime_types = ["image/jpeg", "image/png"]
    buffer = await file.read()
    magic_mime_type = magic.from_buffer(buffer, True)
    m = Magika()
    magika_mime_type = m.identify_bytes(buffer).output.mime_type
    await file.seek(0)
    print(f"file.content_type: {file.content_type}, magic_mime_type: {magic_mime_type}, magika_mime_type: {magika_mime_type}")
    # ファイルが許可されたMIMEタイプの1つであることを確認
    if not (file.content_type in allowed_mime_types and 
            magic_mime_type in allowed_mime_types and
            magika_mime_type in allowed_mime_types):
        raise HTTPException(status_code=400, detail=f"File type not allowed. Allowed types: {', '.join(allowed_mime_types)}")

    return {"filename": file.filename, "content_type": file.content_type, "file_size": file.size}