# Project_folder/app/core/inits.py update
from contextlib import asynccontextmanager
import redis
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from fastapi.middleware.cors import CORSMiddleware
from fastapi_csrf_jinja.middleware import FastAPICSRFJinjaMiddleware
from app.core.settings import STATIC_DIR, MEDIA_DIR, CONFIG, templates
from app.core.database import ASYNC_ENGINE
from app.core.redis import redis_client
from app.utils.commons import to_kst
from app.views import index
@asynccontextmanager
async def lifespan(app: FastAPI):
print("Initializing database......")
# FastAPI 인스턴스 기동시 필요한 작업 수행.
try:
await redis_client.ping() # Redis 연결 테스트
print("Redis connection established......")
except redis.exceptions.ConnectionError:
print("Failed to connect to Redis......")
print("Starting up...")
yield
# FastAPI 인스턴스 종료시 필요한 작업 수행
await redis_client.aclose()
print("Redis connection closed......")
print("Shutting down...")
await ASYNC_ENGINE.dispose()
def including_middleware(app):
app.add_middleware(CORSMiddleware,
allow_origins=["*"], # 실제 프론트 주소
allow_methods=["*"],
allow_headers=["*"],
allow_credentials=True,
max_age=-1)
app.add_middleware(FastAPICSRFJinjaMiddleware,
secret=CONFIG.SECRET_KEY,
cookie_name="csrf_token",
header_name="X-CSRF-Token",)
# swagger를 CSRF_TOKEN적용에서 제외시키는 방법: 라이브러리에서 제공하는 예외 옵션 이름에 맞춰 사용하세요.
# (예: exclude_paths, exempt_paths, exempt_urls 등)
# 이 프로젝트는 swagger로 커스터마이징해서 csrf_token을 적용시켰다.
# exempt_urls=["/swagger/custom/docs", "/swagger/custom/redoc", "/swagger/custom/openapi.json"])
def including_router(app):
app.include_router(index.router, prefix="", tags=["IndexViews"])
def initialize_app():
app = FastAPI(title=CONFIG.APP_NAME,
version=CONFIG.APP_VERSION,
description=CONFIG.APP_DESCRIPTION,
lifespan=lifespan,
docs_url=None, redoc_url=None, openapi_url="/swagger/custom/openapi.json",
)
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
app.mount("/media", StaticFiles(directory=MEDIA_DIR), name="media")
templates.env.globals["STATIC_URL"] = "/static"
templates.env.globals["MEDIA_URL"] = "/media"
templates.env.filters["to_kst"] = to_kst
including_middleware(app)
including_router(app)
return app
# Project_folder/app/core/database.py
import os
from datetime import datetime, timezone
from typing import AsyncGenerator
from sqlalchemy import Integer, DateTime
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import declarative_base, Mapped, mapped_column
from app.core.settings import CONFIG, MEDIA_DIR
PROFILE_IMAGE_UPLOAD_URL = os.path.join(MEDIA_DIR, CONFIG.PROFILE_IMAGE_URL)
ARTICLE_THUMBNAIL_UPLOAD_DIR = os.path.join(MEDIA_DIR, CONFIG.ARTICLE_THUMBNAIL_DIR)
ARTICLE_EDITOR_USER_IMG_UPLOAD_DIR = os.path.join(MEDIA_DIR, CONFIG.ARTICLE_EDITOR_USER_IMG_DIR)
ARTICLE_EDITOR_USER_VIDEO_UPLOAD_DIR = os.path.join(MEDIA_DIR, CONFIG.ARTICLE_EDITOR_USER_VIDEO_DIR)
# 생성할 디렉토리 경로 목록
directory_list = [
PROFILE_IMAGE_UPLOAD_URL,
ARTICLE_THUMBNAIL_UPLOAD_DIR,
ARTICLE_EDITOR_USER_IMG_UPLOAD_DIR,
ARTICLE_EDITOR_USER_VIDEO_UPLOAD_DIR
]
# 목록의 각 경로에 대해 디렉토리 생성
print("3. database: create media subdirectory")
for path in directory_list:
os.makedirs(path, exist_ok=True)
DATABASE_URL = f"{CONFIG.DB_TYPE}+{CONFIG.DB_DRIVER}://{CONFIG.DB_USER}:{CONFIG.DB_PASSWORD}@{CONFIG.DB_HOST}:{CONFIG.DB_PORT}/{CONFIG.DB_NAME}?charset=utf8"
print("4. database: DATABASE_URL ", DATABASE_URL.split("//")[0])
ASYNC_ENGINE = create_async_engine(DATABASE_URL,
echo=CONFIG.DEBUG,
future=True,
pool_size=10, max_overflow=0, pool_recycle=300, # 5분마다 연결 재활용
# encoding="utf-8"
)
# 세션 로컬 클래스 생성
AsyncSessionLocal = async_sessionmaker(
ASYNC_ENGINE,
class_=AsyncSession, # add
expire_on_commit=False,
autocommit=False,
autoflush=False,
# poolclass=NullPool, # SQLite에서는 NullPool 권장
)
"""
autocommit=False 부분은 주의하자.
autocommit=False로 설정하면 데이터를 변경했을때 commit 이라는 사인을 주어야만 실제 저장이 된다.
만약 autocommit=True로 설정할 경우에는 commit이라는 사인이 없어도 즉시 데이터베이스에 변경사항이 적용된다.
그리고 autocommit=False인 경우에는 데이터를 잘못 저장했을 경우 rollback 사인으로 되돌리는 것이 가능하지만
autocommit=True인 경우에는 commit이 필요없는 것처럼 rollback도 동작하지 않는다는 점에 주의해야 한다.
"""
Base = declarative_base() # Base 클래스 (모든 모델이 상속)
class BaseModel(Base):
__abstract__ = True
id: Mapped[int] = mapped_column(Integer, primary_key=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True),
nullable=False,
default=lambda: datetime.now(timezone.utc))
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True),
nullable=False,
default=lambda: datetime.now(timezone.utc),
onupdate=lambda: datetime.now(timezone.utc))
async def get_db() -> AsyncGenerator[AsyncSession, None]:
session: AsyncSession = AsyncSessionLocal()
print(f"[get_session] new session: {id(session)}")
try:
yield session
except Exception as e:
print(f"Session rollback triggered due to exception: {e}")
await session.rollback()
raise
finally:
print(f"[get_session] close session: {id(session)}")
await session.close()
# Project_folder/app/core/redis.py
from redis.asyncio import Redis, ConnectionPool
from app.core.settings import CONFIG
# # Connection Pool 기반 access ##############################
""" # AI Chat # 비동기적으로 ConnectionPool 적용
비동기 Redis 클라이언트와 함께 쓰려면 동기용 ConnectionPool이 아니라 asyncio 전용 풀을 써야 합니다.
즉, redis.ConnectionPool이 아니라 redis.asyncio.ConnectionPool을 사용해야 합니다.
from redis.asyncio import Redis, ConnectionPool
예시 1) asyncio용 ConnectionPool을 직접 생성해서 사용
"""
'''마땅히 설정할 곳이 없어서... 서버 구동시작시 여기를 지나가므로 전역변수처럼 사용'''
ACCESS_COOKIE_MAX_AGE = CONFIG.ACCESS_TOKEN_EXPIRE * 60 # 초 1800 : 30분
CODE_TTL_SECONDS = 10 * 60 # 10분
host = CONFIG.REDIS_HOST if CONFIG.APP_ENV == "production" else "localhost"
port = CONFIG.REDIS_PORT if CONFIG.APP_ENV == "production" else 6379
password = CONFIG.REDIS_PASSWORD if CONFIG.APP_ENV == "production" else None
db = CONFIG.REDIS_DB if CONFIG.APP_ENV == "production" else 0
print("5. redis redis_pool")
redis_pool = ConnectionPool(
host=host,
port=port,
db=db,
password=password,
decode_responses=True, # 문자열 응답을 자동으로 디코딩
max_connections=10)
redis_client = Redis(connection_pool=redis_pool)
# Project_folder/app/core/settings.py update
import os
from functools import lru_cache
from pathlib import Path
from typing import Optional, List
from fastapi.templating import Jinja2Templates
from fastapi_csrf_jinja.jinja_processor import csrf_token_processor
from pydantic import Field, model_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
PRESENT_DIR = os.path.dirname(os.path.abspath(__file__))
APP_DIR = Path(__file__).resolve().parent.parent
ROOT_DIR = Path(__file__).resolve().parent.parent.parent ## root폴더
TEMPLATE_DIR = os.path.join(APP_DIR, 'templates')
STATIC_DIR = os.path.join(APP_DIR, 'static')
MEDIA_DIR = os.path.join(APP_DIR, 'media')
ADMINS = [os.getenv("ADMIN_1"), os.getenv("ADMIN_2")] # 직접 .env파일에서 가져오기
templates = Jinja2Templates(
directory=TEMPLATE_DIR,
context_processors=[csrf_token_processor("csrf_token", "X-CSRF-Token")]
)
class Settings(BaseSettings):
"""여기에서 값이 비어있는 것은 .env 파일에서 채워진다.
그 값은 override 되지 않는다. 흐름상 그럴것 같기는 한데...???
최종적으로 .env에 설정된 값으로 채워져 버리는 것인것 같다."""
APP_ENV: str = "development"
APP_NAME: str = "FastAPI_Jinja"
APP_VERSION: str = "0.0.2"
APP_DESCRIPTION: str = ("FastAPI_Jinja/Javascript을 이용한 프로젝트 개발: \n"
"Jinja Template를 이용하여 Server Side Rendering을 적용하고, \n"
"데이터 입력과정을 pydantic schema적용하고 POST 메서드에서 vanilla javascript를 이용하여 개발 진행")
DEBUG: bool = False
# 기본값은 두지 않고 .env에서 읽히도록 둡니다.
SECRET_KEY: Optional[str] = None
ALGORITHM: str
DB_TYPE: str
DB_DRIVER: str
# 이메일/인증코드
SMTP_FROM: str # = os.getenv("SMTP_FROM", SMTP_USERNAME)
SMTP_USERNAME: str
SMTP_PASSWORD: str
SMTP_PORT: int # = int(os.getenv("SMTP_PORT", 587))
SMTP_HOST: str # = os.getenv("SMTP_HOST", "smtp.gmail.com")
# 액세스/리프페시 토큰
ACCESS_COOKIE_NAME: str
REFRESH_COOKIE_NAME: str
NEW_ACCESS_COOKIE_NAME: str
NEW_REFRESH_COOKIE_NAME: str
ACCESS_TOKEN_EXPIRE: int
REFRESH_TOKEN_EXPIRE: int
# Media Store Path
PROFILE_IMAGE_URL: str
ARTICLE_THUMBNAIL_DIR: str
ARTICLE_EDITOR_USER_IMG_DIR: str
ARTICLE_EDITOR_USER_VIDEO_DIR: str
ORIGINS: List[str] = Field(default_factory=list)
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
extra="ignore",
case_sensitive=False,
)
class DevSettings(Settings):
DEBUG: bool = Field(..., validation_alias="DEBUG_TRUE")
ORIGINS: str = Field(..., validation_alias="DEV_ORIGINS")
DB_NAME: str = Field(..., validation_alias="DEV_DB_NAME")
DB_HOST: str = Field(..., validation_alias="DEV_DB_HOST")
DB_PORT: str = Field(..., validation_alias="DEV_DB_PORT")
DB_USER: str = Field(..., validation_alias="DEV_DB_USER")
DB_PASSWORD: str = Field(..., validation_alias="DEV_DB_PASSWORD")
class ProdSettings(Settings):
APP_NAME: str = Field(..., validation_alias="PROD_APP_NAME")
APP_VERSION: str = Field(..., validation_alias="PROD_APP_VERSION")
APP_DESCRIPTION: str = Field(..., validation_alias="PROD_APP_DESCRIPTION")
ORIGINS: str = Field(..., validation_alias="PROD_ORIGINS")
DB_NAME: str = Field(..., validation_alias="PROD_DB_NAME")
DB_HOST: str = Field(..., validation_alias="PROD_DB_HOST")
DB_PORT: str = Field(..., validation_alias="PROD_DB_PORT")
DB_USER: str = Field(..., validation_alias="PROD_DB_USER")
DB_PASSWORD: str = Field(..., validation_alias="PROD_DB_PASSWORD")
REDIS_HOST: str = Field(..., validation_alias="REDIS_HOST")
REDIS_PORT: str = Field(..., validation_alias="REDIS_PORT")
REDIS_DB: str = Field(..., validation_alias="REDIS_DB")
REDIS_PASSWORD: str = Field(..., validation_alias="REDIS_PASSWORD")
# 운영에서는 SECRET_KEY 필수
@model_validator(mode="after")
def ensure_secret_key(self):
if not self.SECRET_KEY or len(self.SECRET_KEY) < 16:
raise ValueError("In production, SECRET_KEY must be set and sufficiently long.")
return self
@lru_cache(maxsize=1)
def get_settings() -> Settings:
print("1. settings start... Only One")
# model_config의 env_file 설정만으로도 충분하지만, 아래 호출이 있어도 문제는 없습니다.
try:
from dotenv import load_dotenv
load_dotenv(".env", override=False, encoding="utf-8")
except Exception as e:
print(f"load_dotenv error: {e}")
# python-dotenv 미설치 등인 경우에도 설정 로딩은 pydantic-settings가 처리하므로 무시 가능
pass
app_env = os.getenv("APP_ENV", "development").strip().lower()
if app_env == "production":
return ProdSettings()
print("2. settings...APP_ENV: ", app_env)
return DevSettings()
CONFIG = get_settings()
# Project_folder/.env
# 배포시에 주석해제
#APP_ENV=production
DEBUG_TRUE=true
DEBUG_FALSE=false
# production시 DEBUG_FALSE = false
SECRET_KEY=7b6b057992ae6b79d3fecc6d3ebe40626ef4e2ed8221303202e60fdff204d2a7d334b0
ALGORITHM = HS256
DB_TYPE=mysql
DB_DRIVER=aiomysql
DEV_DB_NAME=advanced_db
DEV_DB_HOST=localhost
DEV_DB_PORT=3306
DEV_DB_USER=root
DEV_DB_PASSWORD=<비밀번호>
PROD_APP_NAME=Deployed APP
PROD_APP_VERSION=Deployed V_0
PROD_APP_DESCRIPTION=FastAPI_Jinja를 이용해 개발한 프로젝트에 대한 배포판
PROD_DB_NAME=advanced_db
PROD_DB_HOST=<배포서버 IP주소>
PROD_DB_PORT=<포트번호>
PROD_DB_USER=root
PROD_DB_PASSWORD=<비밀번호>
# redis_config.py 연결 설정: REDIS_PASSWORD None은 에러 유발, 배포 과정에 비번 지정
REDIS_HOST=<배포서버 IP주소>
REDIS_PORT=<포트번호>
REDIS_DB=0
REDIS_PASSWORD=<비밀번호>
# 이메일 보내기 sender 정보
SMTP_USERNAME=abcdefg@gmail.com
SMTP_PASSWORD=iccxyzmnkyfvszux
SMTP_FROM=abcdefg@gmail.com
SMTP_HOST=smtp.gmail.com
SMTP_PORT=587
ACCESS_COOKIE_NAME = access_token
REFRESH_COOKIE_NAME = refresh_token
NEW_ACCESS_COOKIE_NAME = new_access_token
NEW_REFRESH_COOKIE_NAME = new_refresh_token
ACCESS_TOKEN_EXPIRE = 30
REFRESH_TOKEN_EXPIRE = 7
PROFILE_IMAGE_URL = user_images/accounts/profiles
ARTICLE_THUMBNAIL_DIR = user_images/articles/thumbnails
ARTICLE_EDITOR_USER_IMG_DIR = user_images/articles/editor
ARTICLE_EDITOR_USER_VIDEO_DIR = user_videos/articles/editor
# 프론트엔드 개발에 사용되는 포트: 아래 두개(127.0.0.1, localhost)는 별개로 인식한다. 둘다 필요하다.
# 리스트는 한 줄로..., 파싱 가능한 형태로 바꾸기
#DEV_ORIGINS="http://localhost:5173,http://localhost:5100,http://127.0.0.1:5173,http://127.0.0.1:5100"
DEV_ORIGINS=["http://localhost:5173","http://localhost:5174","http://localhost:7100","http://localhost:7101","http://127.0.0.1:5173","http://127.0.0.1:5174","http://127.0.0.1:7100","http://127.0.0.1:7101"]
PROD_ORIGINS=[]
ADMIN_1 = moljin69
ADMIN_2 = test69
# Project_folder/app/models/user.py
from typing import Optional
from sqlalchemy import String, Boolean
from sqlalchemy.orm import Mapped, mapped_column
from app.core.database import BaseModel
class User(BaseModel):
__tablename__ = "users"
# String은 제한 글자수를 지정해야 한다.
username: Mapped[str] = mapped_column(String(20), unique=True, nullable=False)
email: Mapped[str] = mapped_column(String(100), unique=True, nullable=False)
password: Mapped[str] = mapped_column(String(255), nullable=False)
img_path: Mapped[Optional[str]] = mapped_column(String(200), nullable=True)
is_admin: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
"""
- unique 와 index 중복
- unique=True만으로도 고유 인덱스가 생성됩니다. index=True는 중복이므로 제거해도 됩니다(권장).
- 비밀번호 길이
- password는 해시를 저장하므로 길이를 넉넉히 두는 것이 안전합니다(예: String(255)). 해시 종류에 따라 100자를 넘길 수 있습니다.
- 기본 키/타입
- id는 Integer(primary_key=True)면 충분합니다. 아주 큰 규모를 예상하면 BigInteger도 고려할 수 있습니다.
"""
# project_folder/app/models/article.py
from datetime import datetime, timezone
from typing import Optional
from sqlalchemy import DateTime, ForeignKey, Integer, String, func, Text
from sqlalchemy.orm import relationship, Mapped, mapped_column, backref
from app.core.database import Base
class Article(Base):
__tablename__ = "articles"
id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
# String은 제한 글자수를 지정해야 한다.
title: Mapped[str] = mapped_column(String(100), index=True)
img_path: Mapped[Optional[str]] = mapped_column(String(200), nullable=True)
content: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc))
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
# users.id에서 users는 테이블명
# 외래키를 사용할 때, 제약 조건에 name을 ForeignKey 안에 ForeignKey("users.id", name="fk_author_id") 이렇게 넣어라.
author_id: Mapped[int] = mapped_column(Integer, ForeignKey("users.id", name="fk_author_id", ondelete='CASCADE'), nullable=False)
# author_id가 nullable=True 이므로 Optional["User"]가 일관됩니다.
author: Mapped["User"] = relationship("User", backref=backref("article_user_set",
lazy="selectin",
cascade="all, delete-orphan",
passive_deletes=True), lazy="selectin")
""" author 속성은 User 모델에서 Article 모델을 참조하기 위해 추가했다. 위와 같이 relationship으로 author(User) 속성을 생성하면,
게시글 객체(예: article)에서 연결된 저자의 username 을 article.user.username 처럼 참조할 수 있다.
relationship의 첫 번째 파라미터는 참조할 모델명이고 두 번째 backref 파라미터는 역참조 설정이다. 역참조란 쉽게 말해 User 에서 Article을 거꾸로 참조하는 것을 의미한다.
한 User에는 여러 개의 Article이 생성 수 있는데 역참조는 이 User가 작성한 Article 들을 참조할 수 있게 한다.
예를 들어 어떤 User에 해당하는 객체가 a_user 라면 a_user.article_user_set 같은 코드로 해당 User가 작성한 Article 들을 참조할 수 있다.
"""
def __repr__(self):
return f"<Article(id={self.id}, title='{self.title}', author_id={self.author_id}, created_at={self.created_at})>"
"""
현재 템플릿 단에서 `<div class="uk-text-right">{{ article.author }}</div>` 여기에서
sqlalchemy.exc.StatementError: (sqlalchemy.exc.MissingGreenlet) greenlet_spawn has not been called; can't call await_only() here. Was IO attempted in an unexpected place?
[SQL: SELECT users.id AS users_id, users.username AS users_username, users.email AS users_email, users.password AS users_password, users.created_at AS users_created_at
FROM users
WHERE users.id = %s]
[parameters: [{'pk_1': 21}]]
에러가 발생한다. 해결방법은? AI Chat
해당 에러는 템플릿에서 article.author에 접근하는 순간, SQLAlchemy가 “지연 로딩(lazy load)”을 수행하려고 하며,
비동기 드라이버(aiomysql)를 쓰는 환경에서 Jinja2의 동기 렌더링 컨텍스트와 충돌해서 발생합니다.
즉, 템플릿 렌더링 중에 DB I/O가 필요해져서 MissingGreenlet가 납니다.
1. (선택) 모델 관계 기본 전략을 selectin으로 설정
- 매 쿼리마다 options를 달기 어렵다면, 관계 정의에서 lazy="selectin"으로 설정해 지연 로딩 대신 배치 로딩을 기본으로 사용합니다.
class Article(Base):
# ...
author = relationship("User", back_populates="articles", lazy="selectin")
"""
# project_folder/app/utils/commons.py
import datetime
import random
import re
import shutil
import uuid
from typing import LiteralString
from email_validator import validate_email, EmailNotValidError
from fastapi import status, UploadFile, HTTPException, Request
import os
import aiofiles as aio
from pydantic_core import PydanticCustomError
from starlette.concurrency import run_in_threadpool
from app.core.settings import MEDIA_DIR, CONFIG, templates
from app.models.user import User
try:
from zoneinfo import ZoneInfo # Python 3.9+
KST = ZoneInfo("Asia/Seoul")
except Exception as e:
print("6. utils/commons: zoneInfo error ", e)
# tzdata가 없을 때를 위한 안전한 폴백(고정 +09:00)
KST = datetime.timezone(datetime.timedelta(hours=9), name="KST")
def get_times(): # 시간 표시는 전역변수로 두지 말고, 매번 불러서 렌더링해야한다.
# 1) 항상 UTC aware부터 시작
_NOW_TIME_UTC= datetime.datetime.now(datetime.timezone.utc)
# 2) 필요한 지역시간(KST)으로 변환
_NOW_TIME = _NOW_TIME_UTC.astimezone(KST)
return _NOW_TIME_UTC, _NOW_TIME
def to_kst(dt: datetime.datetime | None, fmt: str = "%Y-%m-%d %H:%M:%S") -> str:
"""
UTC(또는 타임존 정보가 있는) datetime을 KST로 변환해 문자열로 반환합니다.
- dt가 naive(타임존 없음)이면 UTC로 간주합니다.
- dt가 None이면 빈 문자열을 반환합니다.
- fmt로 출력 포맷을 지정할 수 있습니다.
"""
if dt is None:
return ""
if dt.tzinfo is None:
dt = dt.replace(tzinfo=datetime.timezone.utc)
return dt.astimezone(KST).strftime(fmt)
# 로컬파트/도메인 유효성 정규식
EMAIL_LOCAL_RE = re.compile(
r"^(?![.])(?!.*[.]{2})[A-Za-z0-9!#$%&'*+/=?^_`{|}~-]+(?:\.[A-Za-z0-9!#$%&'*+/=?^_`{|}~-]+)*(?<!\.)$"
)
DOMAIN_LABEL_RE = re.compile(r"^(?!-)[A-Za-z0-9-]{1,63}(?<!-)$")
TLD_ALPHA_RE = re.compile(r"^[A-Za-z]{2,24}$")
INVALID_EMAIL: LiteralString = "올바른 이메일 형식이 아닙니다."
def strict_email(v):
if v is None:
print(f"Email Strict Validation Error: ", '빈 값은 허용되지 않습니다.')
raise PydanticCustomError('empty_value', "올바른 이메일 형식이 아닙니다.")
s = str(v).strip()
if not s:
print(f"Email Strict Validation Error: ", '빈 값은 허용되지 않습니다.')
raise PydanticCustomError('empty_value', INVALID_EMAIL)
if s.count('@') != 1:
print(f"Email Strict Validation Error: ", '올바른 이메일 형식이 아닙니다.')
raise PydanticCustomError('invalid_email', INVALID_EMAIL)
local, domain = s.split('@', 1)
# 길이 제한
if len(s) > 254:
print(f"Email Strict Validation Error: ", '이메일 전체 길이가 너무 깁니다(최대 254).')
raise PydanticCustomError('invalid_email', INVALID_EMAIL)
if len(local) > 64:
print(f"Email Strict Validation Error: ", '로컬파트 길이가 너무 깁니다(최대 64).')
raise PydanticCustomError('invalid_email', INVALID_EMAIL)
# 로컬파트(dot-atom) 검증
if not EMAIL_LOCAL_RE.fullmatch(local):
print(f"Email Strict Validation Error: ", '허용되지 않는 이메일 로컬파트입니다.')
raise PydanticCustomError('invalid_email', INVALID_EMAIL)
# 도메인 정규화 및 검증
domain = domain.lower().rstrip('.') # 끝의 점(FQDN 표기) 제거
if not domain:
print(f"Email Strict Validation Error: ", '올바른 이메일 형식이 아닙니다.')
raise PydanticCustomError('invalid_email', INVALID_EMAIL)
if len(domain) > 253:
print(f"Email Strict Validation Error: ", '도메인 길이가 너무 깁니다(최대 253).')
raise PydanticCustomError('invalid_email', INVALID_EMAIL)
labels = domain.split('.')
if len(labels) < 2:
print(f"Email Strict Validation Error: ", '도메인에 점(.)이 최소 1개 포함되어야 합니다.')
raise PydanticCustomError('invalid_email', INVALID_EMAIL)
for label in labels:
if not DOMAIN_LABEL_RE.fullmatch(label):
print(f"Email Strict Validation Error: ", '허용되지 않는 도메인 라벨이 포함되어 있습니다.')
raise PydanticCustomError('invalid_email', INVALID_EMAIL)
tld = labels[-1]
# TLD: 영문 2~24자 또는 punycode(xn--)
if tld.startswith('xn--'):
if not (5 <= len(tld) <= 63):
print(f"Email Strict Validation Error: ", '유효하지 않은 최상위 도메인입니다.')
raise PydanticCustomError('invalid_email', INVALID_EMAIL)
else:
if not TLD_ALPHA_RE.fullmatch(tld):
print(f"Email Strict Validation Error: ", '유효하지 않은 최상위 도메인입니다.')
raise PydanticCustomError('invalid_email', INVALID_EMAIL)
if tld.isdigit():
print(f"Email Strict Validation Error: ", '유효하지 않은 최상위 도메인입니다.')
raise PydanticCustomError('invalid_email', INVALID_EMAIL)
# 정상: 정규화(도메인은 소문자)
return f"{local}@{domain}"
async def random_string(length:int, _type:str):
if _type == "full":
string_pool = "0za1qw2sc3de4rf5vb6gt7yh8nm9juiklop"
elif _type == "string":
string_pool = "abcdefghijklmnopqrstuvwxyz"
else:
string_pool = "0123456789"
result = random.choices(string_pool, k=length)
seperator = ""
return seperator.join(result)
async def file_renaming(username:str, ext:str):
date = datetime.datetime.now().strftime("%Y%m%d_%H%M_%S%f")
random_str = await random_string(8, "full")
new_filename = f"{username}_{date}{random_str}"
return f"{new_filename}{ext}"
async def upload_single_image(path:str, user: User, imagefile: UploadFile = None):
try:
upload_dir = f"{path}"+"/"+f"{user.id}"+"/" # d/t Linux
print("upload_dir: ", upload_dir)
url = await file_write_return_url(upload_dir, user, imagefile, "media", _type="image")
return url
except Exception as e:
print(e)
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="이미지 파일이 제대로 Upload되지 않았습니다. ")
async def file_write_return_url(upload_dir: str, user: User, file: UploadFile, _dir: str, _type: str):
if not os.path.exists(upload_dir):
os.makedirs(upload_dir)
filename_only, ext = os.path.splitext(file.filename)
if len(file.filename.strip()) > 0 and len(filename_only) > 0:
upload_filename = await file_renaming(user.username, ext)
file_path = upload_dir + upload_filename
async with aio.open(file_path, "wb") as outfile:
if _type == "image":
_CHUNK = 1024 * 1024
while True:
chunk = await file.read(_CHUNK)
if not chunk:
break
await outfile.write(chunk)
elif _type == "video":
# await outfile.write(await file.read())
# 대용량 업로드: 청크 단위로 읽어서 바로 쓰기 (메모리 사용 최소화)
_CHUNK = 8 * 1024 * 1024 # 8MB
while True:
chunk = await file.read(_CHUNK)
if not chunk:
break
await outfile.write(chunk)
await outfile.flush()
url = file_path.split(_dir)[1]
return url
else:
return None
"""# 존재 체크-후-삭제 사이에 경쟁 상태가 생길 수 있슴,
권장: 존재 여부 체크 없이 바로 삭제 시도 (경쟁 상태 방지)"""
async def remove_file_path(path:str): # 파일 삭제
try:
await run_in_threadpool(os.remove, path)
except FileNotFoundError:
pass # 이미 없으면 무시
async def remove_empty_dir(_dir:str): # 빈 폴더 삭제
try:
await run_in_threadpool(os.rmdir, _dir) # 비어 있을 때만 성공
except FileNotFoundError:
pass # 이미 사라졌다면 무시
except OSError as e:
print("비어있지 않거나 잠겨 있는 경우: ", e)
pass # 비어있지 않거나 잠겨 있으면 무시(필요 시 로깅)
async def remove_dir_with_files(_dir:str): # 파일을 포함하여 폴더 삭제
try:
await run_in_threadpool(shutil.rmtree, _dir)
# 이미지 저장 디렉토리 및 파일을 삭제, ignore_errors, onerror 파라미터 넣지 않아도 작동한다.
except FileNotFoundError:
pass
async def old_image_remove(filename:str, path:str):
try:
filename_only, ext = os.path.splitext(filename)
if len(filename.strip()) > 0 and len(filename_only) > 0 and path is not None:
# old_image_path = f'{APP_DIR}{path}' # \\없어도 된다. url 맨 앞에 \\ 있다.
old_image_path = f'{MEDIA_DIR}'+'/'+f'{path}'
await remove_file_path(old_image_path)
except Exception as e:
print(e)
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="이미지 파일이 제대로 Upload되지 않았습니다. ")
# ----------------- 유틸: 이메일 유효성 체크 -----------------
def is_valid_email(email: str) -> bool:
try:
validate_email(email)
return True
except EmailNotValidError:
return False
def is_https(request: Request) -> bool:
forwarded = request.headers.get("x-forwarded-proto")
scheme = (forwarded or request.url.scheme or "").lower()
return scheme == "https"
def refresh_expire():
return datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(days=CONFIG.REFRESH_TOKEN_EXPIRE)
async def render_with_times(request: Request, template: str, extra_context: dict | None = None):
now_time_utc, now_time = get_times()
context = {
"request": request,
"now_time_utc": now_time_utc.strftime('%Y-%m-%d %H:%M:%S.%f'),
"now_time": now_time.strftime('%Y-%m-%d %H:%M:%S.%f'),
}
if extra_context:
context.update(extra_context)
return templates.TemplateResponse(template, context)
def create_orm_id(objs_all, user):
"""비동기 함수로 바꿔야 하나???"""
try:
unique_num = str(objs_all[0].id + 1) # 고유해지지만, model.id와 일치하지는 않는다. 삭제된 놈들이 있으면...
print("unique_num:::::::::::::: ", unique_num,)
except Exception as e:
print("c_orm_id Exception error::::::::: 임의로 1로... 할당 ", e)
unique_num = str(1) # obj가 첫번째 것인 경우: 임의로 1로... 할당
_random_string = str(uuid.uuid4())
username = user.username
orm_id = unique_num + ":" + username + "_" + _random_string
return orm_id
pip install alembic
alembic init miagrations
# Project_folder/alembic,ini # sqlalchemy.url = 공란으로 수정
sqlalchemy.url =
# Project_folder/app/migrations/env.py
import asyncio
from logging.config import fileConfig
import os
import sys
from sqlalchemy import pool
from sqlalchemy.ext.asyncio import async_engine_from_config
from alembic import context
sys.path.insert(0, os.path.realpath(os.path.join(os.path.dirname(__file__), "..")))
from app.core.database import Base, DATABASE_URL
# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config
# Interpret the config file for Python logging.
# This line sets up loggers basically.
# if config.config_file_name is not None:
# fileConfig(config.config_file_name)
if config.config_file_name is not None:
try:
fileConfig(config.config_file_name, encoding='utf-8')
except TypeError:
# older versions of fileConfig don't support the encoding parameter
fileConfig(config.config_file_name)
# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
''' 반드시 모델들을 임포트 하라. '''
from app.models import user
# from app.test import exam
# target_metadata = mymodel.Base.metadata
# target_metadata = None
target_metadata = Base.metadata
# other values from the config, defined by the needs of env.py,
# can be acquired:
# my_important_option = config.get_main_option("my_important_option")
# ... etc.
def run_migrations_offline() -> None:
"""Run migrations in 'offline' mode.
This configures the context with just a URL
and not an Engine, though an Engine is acceptable
here as well. By skipping the Engine creation
we don't even need a DBAPI to be available.
Calls to context.execute() here emit the given string to the
script output.
"""
url = config.get_main_option("sqlalchemy.url")
context.configure(
url=url,
target_metadata=target_metadata,
literal_binds=True,
dialect_opts={"paramstyle": "named"},
)
with context.begin_transaction():
context.run_migrations()
def do_run_migrations(connection):
context.configure(connection=connection, target_metadata=target_metadata)
with context.begin_transaction():
context.run_migrations()
async def run_migrations_online() -> None:
"""Run migrations in 'online' mode for an async application."""
# 환경 변수가 설정되지 않았다면 alembic.ini의 기본값을 사용하도록 폴백 설정 (선택 사항)
db_url = DATABASE_URL
if not db_url:
db_url = config.get_main_option("sqlalchemy.url")
if not db_url:
raise ValueError("데이터베이스 URL을 찾을 수 없습니다. .env 파일이나 환경 변수를 확인하세요.")
connectable = async_engine_from_config(
config.get_section(config.config_ini_section, {}),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
future=True, # SQLAlchemy 2.0
url=db_url # 여기에 환경 변수에서 가져온 URL을 전달 (개발 및 배포모드 모두 커버)
)
async with connectable.connect() as connection:
await connection.run_sync(do_run_migrations)
await connectable.dispose()
if context.is_offline_mode():
run_migrations_offline()
else:
asyncio.run(run_migrations_online())
alembic revision --autogenerate -m "User image add"
alembic upgrade head
# project_folder/app/views/index.py
from fastapi import APIRouter, Request
from fastapi.openapi.docs import get_redoc_html
from starlette.responses import HTMLResponse
from app.core.settings import templates
router = APIRouter()
@router.get("/")
def index(request: Request,):
template = "commons/index.html"
context = {"request": request, "message": "Hello World"}
return templates.TemplateResponse(template,context)
SWAGGER_JS = "https://cdn.jsdelivr.net/npm/swagger-ui-dist@5/swagger-ui-bundle.js"
SWAGGER_CSS = "https://cdn.jsdelivr.net/npm/swagger-ui-dist@5/swagger-ui.css"
SWAGGER_FAV = "https://fastapi.tiangolo.com/img/favicon.png"
COOKIE_NAME = "csrf_token"
HEADER_NAME = "X-CSRF-Token"
open_api_url = "/swagger/custom/openapi.json"
@router.get("/swagger/custom/docs", include_in_schema=False)
async def custom_docs():
"""여기에 html 문서자체를 커스텀 마이징하는 로직을 추가하면 된다.
CSRF_TOKEN을 주입하는 경우 등...
csrf_token을 header에 추가하는 커스텀 Swagger UI 페이지를 반환합니다.
"""
from main import app
openapi_url = app.openapi_url or open_api_url
html = f"""<!doctype html>
<html>
<head>
<meta charset="utf-8" />
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Swagger Custom Docs</title>
<link rel="stylesheet" href="{SWAGGER_CSS}" />
<link rel="icon" type="image/png" href="{SWAGGER_FAV}" />
<style>
html {{ box-sizing: border-box; overflow: -moz-scrollbars-vertical; overflow-y: scroll; }}
*, *:before, *:after {{ box-sizing: inherit; }}
body {{ margin:0; background: #fafafa; }}
</style>
</head>
<body>
<div id="swagger-ui"></div>
<script src="{SWAGGER_JS}"></script>
<script>
window.onload = function() {{
// 쿠키에서 값 읽는 유틸 (안정적인 방법)
function getCookie(name) {{
const m = document.cookie.match('(^|;)\\s*' + name + '\\s*=\\s*([^;]+)');
return m ? decodeURIComponent(m[2]) : null;
}}
// Swagger UI 초기화
const ui = SwaggerUIBundle({{
url: "{openapi_url}",
dom_id: '#swagger-ui',
deepLinking: true,
presets: [SwaggerUIBundle.presets.apis],
layout: "BaseLayout",
// requestInterceptor - 모든 요청이 전송되기 전에 호출됩니다.
requestInterceptor: function (req) {{
try {{
// 안전한 메서드에는 CSRF를 추가하지 않습니다.
const safeMethods = ['GET','HEAD','OPTIONS','TRACE'];
if (!safeMethods.includes((req.method || '').toUpperCase())) {{
const csrf = getCookie("{COOKIE_NAME}");
if (csrf) {{
// 헤더 추가 (대소문자 무관)
req.headers["{HEADER_NAME}"] = csrf;
}}
// 쿠키 전송이 필요하면 credentials 설정 (CORS 허용 필요)
// 'include' 는 cross-site에서도 쿠키를 전송합니다. same-origin이면 'same-origin' 사용 가능.
req.credentials = 'include';
}}
}} catch (e) {{
console.warn('requestInterceptor error:', e);
}}
return req;
}},
}});
window.ui = ui;
}};
</script>
</body>
</html>"""
# return get_swagger_ui_html(
# openapi_url=open_api_url, # openapi 스펙 파일 URL
# title="Custom API Docs",
# )
return HTMLResponse(html)
@router.get("/swagger/custom/redoc", include_in_schema=False)
async def custom_redoc():
"""여기에 html 문서자체를 커스텀 마이징하는 로직을 추가하면 된다."""
return get_redoc_html(
openapi_url=open_api_url, # 스펙 파일 경로
title="Custom ReDoc", # 페이지 제목
redoc_favicon_url="https://fastapi.tiangolo.com/img/favicon.png",
)
