Python

파이썬 가상환경

• Colab: 구글이 무료로 제공하는 주피터 노트북(Jupyter Notebook) 환경
• Anaconda: 주피터 노트북 환경을 로컬에 실행하기 위한 프로그램
• virtualenv, pip, venv, pyenv, pipenv...
• Poetry: Python packaging and dependency management made easy
• PDM, Rye...
• uv: Python packaging in Rust

# Install uv on macOS.
$ curl -LsSf https://astral.sh/uv/install.sh | sh

# Use a specific Python version in the current directory:
$ uv python pin 3.12
Pinned `.python-version` to `3.12`

Zsh autocompletion in mac

$ echo 'eval "$(uv generate-shell-completion zsh)"' >> ~/.zshrc

파이썬 SQL

uv add databases aiosqlite sqlalchemy

• databases
• aiosqlite
• sqlalchemy

db.py

import sqlalchemy
from sqlalchemy import MetaData
from databases import Database

database = Database('sqlite+aiosqlite:///local.db')
engine = sqlalchemy.create_engine(str(database.url).replace("+aiosqlite", ''), echo=True)
metadata = MetaData()
metadata.create_all(engine)

models.py

import sqlalchemy
from sqlalchemy import Table, Column, Integer, String, DateTime, Text
from db import metadata

request_logs = Table(
    "request_logs",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("method", String, nullable=False),
    Column("path", String, nullable=False),
    Column("request_headers", Text, nullable=False),
    Column("request_params", Text, nullable=True),
    Column("request_body", Text, nullable=True),
    Column("client_ip", String, nullable=False),
    Column("server_ip", String, nullable=False),
    Column("response_status", Integer, nullable=False),
    Column("response_headers", Text, nullable=False),
    Column("response_body", Text, nullable=True),
    Column("created_at", DateTime, server_default=sqlalchemy.text("CURRENT_TIMESTAMP")),
)

main.py

from fastapi import FastAPI, Request, Response
from contextlib import asynccontextmanager
from db import database
from models import request_logs
import json
import socket

server_ip = socket.gethostbyname(socket.gethostname())

@asynccontextmanager
async def lifespan(app: FastAPI):
    await database.connect()
    yield
    await database.disconnect()


app = FastAPI(lifespan=lifespan)


@app.middleware("http")
async def log_request(request: Request, call_next):
    try:
        request_params = await request.json()
    except Exception:
        request_params = dict(request.query_params)
    request_body = await request.body()

    response = await call_next(request)

    chunks = []
    async for chunk in response.body_iterator:
        chunks.append(chunk)
    response_body = b"".join(chunks)

    log_data = {
        "method": request.method,
        "path": request.url.path,
        "request_headers": json.dumps(dict(request.headers)),
        "request_params": json.dumps(request_params),
        "request_body": request_body.decode("utf-8", errors="ignore"),
        "client_ip": request.client.host,
        "server_ip": server_ip,
        "response_status": response.status_code,
        "response_headers": json.dumps(dict(response.headers)),
        "response_body": response_body.decode("utf-8", errors="ignore"),
    }

    await database.execute(request_logs.insert().values(**log_data))

    return Response(
        content=response_body,
        status_code=response.status_code,
        headers=dict(response.headers),
        media_type=response.media_type,
    )


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

db_async.py

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base

DB_URL = 'sqlite+aiosqlite:///local.db'

Base = declarative_base()
engine = create_async_engine(DB_URL, echo=True)
SessionLocal = sessionmaker(
    bind=engine,
    autocommit=False,
    autoflush=False,
    expire_on_commit=False,
    class_=AsyncSession,
)


async def init_db():
    async with engine.connect() as conn:
        await conn.run_sync(Base.metadata.create_all)  # NOTE: 동기적 호출

테이블을 정의할 때는 databases를 이용할 수 없으므로 SQLAlchemy를 직접 사용해야한다. 다만, create_all 함수는 동기적 호출을 요구하므로 create_async_engine 와 sessionmaker 를 사용하는 코드를 굳이 작성할 필요는 없다.