WTMO-dev
AI
CV
basic
DL
basic
ML
basic
NLP
basic
dspy
hugging face
llama index
ollama
C++
basic
COMPUTER SCIENCE
basic
ETC
GIT
Blog
basic
IDE
VSC
python
setting
JAVA
basic
MATH
Basic
Statistic
NETWORK
basic
OS
Window
WSL
basic
basic
PROJECT
PYTHON
Advance
Basic
Framework
Django
FastAPI
Library
Module
Home
PROJECT PRACTICE
Crawling
EDA
Kaggle
Python
CV
Contact
Copyright © 2024
WTMO-dev
Home
>
PYTHON
>
Framework
> FastAPI
Now Loading ...
FastAPI
FastAPI Advance
Advance 준비중입니다. Advance
PYTHON
/
Framework
/
FastAPI
· 2024-02-08
FastAPI SQL
SQL SQL은 다음과 같이 구조화 하여 사용할 수 있습니다. sql_app/__init__.py: empty sql_app/database.py sql_app/models.py sql_app/schemas.py sql_app/crud.py sql_app/main.py sql_app/database.py database 파일은 다음과 같이 구성됩니다. {% highlight python %} from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker SQLALCHEMY_DATABASE_URL = “sqlite:///./sql_app.db” SQLALCHEMY_DATABASE_URL = “postgresql://user:password@postgresserver/db” engine = create_engine( SQLALCHEMY_DATABASE_URL, connect_args={“check_same_thread”: False} ) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base() {% endhighlight %} sql_app/models.py models 파일은 다음과 같이 구성됩니다. {% highlight python %} from sqlalchemy import Boolean, Column, ForeignKey, Integer, String from sqlalchemy.orm import relationship from .database import Base class User(Base): tablename = “users” id = Column(Integer, primary_key=True) email = Column(String, unique=True, index=True) hashed_password = Column(String) is_active = Column(Boolean, default=True) items = relationship("Item", back_populates="owner") class Item(Base): tablename = “items” id = Column(Integer, primary_key=True) title = Column(String, index=True) description = Column(String, index=True) owner_id = Column(Integer, ForeignKey("users.id")) owner = relationship("User", back_populates="items") {% endhighlight %} sql_app/schemas.py schemas 파일은 다음과 같이 구성됩니다. {% highlight python %} from typing import Union from pydantic import BaseModel class ItemBase(BaseModel): title: str description: Union[str, None] = None class ItemCreate(ItemBase): pass class Item(ItemBase): id: int owner_id: int class Config: orm_mode = True class UserBase(BaseModel): email: str class UserCreate(UserBase): password: str class User(UserBase): id: int is_active: bool items: list[Item] = [] class Config: orm_mode = True {% endhighlight %} sql_app/crud.py crud 파일은 다음과 같이 구성됩니다. {% highlight python %} from sqlalchemy.orm import Session from . import models, schemas def get_user(db: Session, user_id: int): return db.query(models.User).filter(models.User.id == user_id).first() def get_user_by_email(db: Session, email: str): return db.query(models.User).filter(models.User.email == email).first() def get_users(db: Session, skip: int = 0, limit: int = 100): return db.query(models.User).offset(skip).limit(limit).all() def create_user(db: Session, user: schemas.UserCreate): fake_hashed_password = user.password + “notreallyhashed” db_user = models.User(email=user.email, hashed_password=fake_hashed_password) db.add(db_user) db.commit() db.refresh(db_user) return db_user def get_items(db: Session, skip: int = 0, limit: int = 100): return db.query(models.Item).offset(skip).limit(limit).all() def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int): db_item = models.Item(**item.dict(), owner_id=user_id) db.add(db_item) db.commit() db.refresh(db_item) return db_item {% endhighlight %} sql_app/main.py main 파일은 다음과 같이 구성됩니다. {% highlight python %} from fastapi import Depends, FastAPI, HTTPException from sqlalchemy.orm import Session from . import crud, models, schemas from .database import SessionLocal, engine models.Base.metadata.create_all(bind=engine) app = FastAPI() Dependency def get_db(): db = SessionLocal() try: yield db finally: db.close() @app.post(“/users/”, response_model=schemas.User) def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)): db_user = crud.get_user_by_email(db, email=user.email) if db_user: raise HTTPException(status_code=400, detail=”Email already registered”) return crud.create_user(db=db, user=user) @app.get(“/users/”, response_model=list[schemas.User]) def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): users = crud.get_users(db, skip=skip, limit=limit) return users @app.get(“/users/{user_id}”, response_model=schemas.User) def read_user(user_id: int, db: Session = Depends(get_db)): db_user = crud.get_user(db, user_id=user_id) if db_user is None: raise HTTPException(status_code=404, detail=”User not found”) return db_user @app.post(“/users/{user_id}/items/”, response_model=schemas.Item) def create_item_for_user( user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db) ): return crud.create_user_item(db=db, item=item, user_id=user_id) @app.get(“/items/”, response_model=list[schemas.Item]) def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): items = crud.get_items(db, skip=skip, limit=limit) return items {% endhighlight %} W. middleware 다음은 미들웨어를 사용한 예시를 제시합니다. {% highlight python %} @app.middleware(“http”) async def db_session_middleware(request: Request, call_next): response = Response(“Internal server error”, status_code=500) try: request.state.db = SessionLocal() response = await call_next(request) finally: request.state.db.close() return response Dependency def get_db(request: Request): return request.state.db {% endhighlight %}
PYTHON
/
Framework
/
FastAPI
· 2024-02-07
FastAPI types
Enum Enum 클래스는 다음과 같이 사용할 수 있습니다. {% highlight python %} from enum import Enum class ModelName(str, Enum): alexnet = “alexnet” resnet = “resnet” lenet = “lenet” {% endhighlight %} 값을 사용하는 방식은 다음과 같이 사용 가능합니다. model.value Model.<key> {% highlight python %} app = FastAPI() @app.get(“/models/{model_name}”) async def get_model(model_name: ModelName): if model_name is ModelName.alexnet: {% endhighlight %} Optional Optional type의 경우 None일 수 있는 데이터를 의미하며 다음과 같이 작성할 수 있습니다. {% highlight python %} from typing import Optional app = FastAPI() @app.get(“/items/{item_id}”) async def read_item(item_id: Optional[int]): return {“item_id”: item_id} {% endhighlight %} httpUrl httpUrl type의 경우 url형식을 받을 수 있는 형식입니다. {% highlight python %} from fastapi import FastAPI from pydantic import BaseModel, HttpUrl app = FastAPI() class Image(BaseModel): url: HttpUrl name: str @app.put(“/items”) async def update_item(image: Image): results = {“image”: image} return results {% endhighlight %} output type output type의 경우 일반적인 방식으로 활용이 가능합니다. response_model=None은 2종류 이상의 타입이 가능할경우 pydantic의 규제를 피할때 사용할 수 있습니다. response_model_exclude_unset=True은 output을 넘겨줄때 pydantic의 default value는 무시하는 방법입니다. response_model_exclude={""}은 output을 넘겨줄때 넘겨주지 않을 value를 지정하는 방법입니다. {% highlight python %} from fastapi import FastAPI, Response from fastapi.responses import RedirectResponse app = FastAPI() @app.get(“/portal”, response_model=None, response_model_exclude_unset=True, response_model_exclude={“tax”}) async def get_portal(teleport: bool = False) -> Response | dict: if teleport: return RedirectResponse(url=”https://www.youtube.com/watch?v=dQw4w9WgXcQ”) return {“message”: “Here’s your interdimensional portal.”} {% endhighlight %} and others 이외에도 다양한 data type이 존재하며 다음과 같습니다. datetime.datetime datetime.date datetime.time datetime.timedelta frozenset bytes Decimal schema examples pydantic으로 형식을 만들때는 다음과 같이 예시문을 만들 수 있습니다. 예시문을 활용하면 docs에서 예시문을 확인할 수 있습니다. {% highlight python %} class Item(BaseModel): name: str description: str | None = None price: float tax: float | None = None model_config = { "json_schema_extra": { "examples": [ { "name": "Foo", "description": "A very nice Item", "price": 35.4, "tax": 3.2, } ] } } {% endhighlight %} pydantic이 아닐경우는 Body, Query, …에서 직접 예시문을 작성할 수도 있습니다. {% highlight python %} Body( examples=[ { “name”: “Foo”, “description”: “A very nice Item”, “price”: 35.4, “tax”: 3.2, } ], ), {% endhighlight %}
PYTHON
/
Framework
/
FastAPI
· 2024-02-06
FastAPI intro Num.3
Security API를 만드는데 있어서 보안사항은 매우 중요합니다. 아래의 순서를 따라 보안을 설정할 수 있습니다. {% highlight shell %} pip install python-jose # jwt token module pip install passlib[bcrypt] # password hashing module {% endhighlight %} 만약 (trapped) error reading bcrypt version과 같은 문제가 발생하면 아래의 코드를 작동시켜야 합니다. {% highlight shell %} pip install bcrypt==4.0.1 {% endhighlight %} 아래의 방법은 보안을 설정하는 방법들을 제안합니다. {% highlight python %} from datetime import datetime, timedelta, timezone from typing import Annotated, Union from fastapi import Depends, FastAPI, HTTPException, status from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from jose import JWTError, jwt from passlib.context import CryptContext from pydantic import BaseModel to get a string like this run: openssl rand -hex 32 SECRET_KEY = “09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7” ALGORITHM = “HS256” ACCESS_TOKEN_EXPIRE_MINUTES = 30 fake_users_db = { “johndoe”: { “username”: “johndoe”, “full_name”: “John Doe”, “email”: “johndoe@example.com”, “hashed_password”: “$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW”, “disabled”: False, } } class Token(BaseModel): access_token: str token_type: str class TokenData(BaseModel): username: Union[str, None] = None class User(BaseModel): username: str email: Union[str, None] = None full_name: Union[str, None] = None disabled: Union[bool, None] = None class UserInDB(User): hashed_password: str pwd_context = CryptContext(schemes=[“bcrypt”], deprecated=”auto”) oauth2_scheme = OAuth2PasswordBearer(tokenUrl=”token”) app = FastAPI() def verify_password(plain_password, hashed_password): return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password): return pwd_context.hash(password) def get_user(db, username: str): if username in db: user_dict = db[username] return UserInDB(**user_dict) def authenticate_user(fake_db, username: str, password: str): user = get_user(fake_db, username) if not user: return False if not verify_password(password, user.hashed_password): return False return user def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None): to_encode = data.copy() if expires_delta: expire = datetime.now(timezone.utc) + expires_delta else: expire = datetime.now(timezone.utc) + timedelta(minutes=15) to_encode.update({“exp”: expire}) encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]): credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail=”Could not validate credentials”, headers={“WWW-Authenticate”: “Bearer”}, ) try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) username: str = payload.get(“sub”) if username is None: raise credentials_exception token_data = TokenData(username=username) except JWTError: raise credentials_exception user = get_user(fake_users_db, username=token_data.username) if user is None: raise credentials_exception return user async def get_current_active_user( current_user: Annotated[User, Depends(get_current_user)], ): if current_user.disabled: raise HTTPException(status_code=400, detail=”Inactive user”) return current_user @app.post(“/token”) async def login_for_access_token( form_data: Annotated[OAuth2PasswordRequestForm, Depends()], ) -> Token: user = authenticate_user(fake_users_db, form_data.username, form_data.password) if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail=”Incorrect username or password”, headers={“WWW-Authenticate”: “Bearer”}, ) access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) access_token = create_access_token( data={“sub”: user.username}, expires_delta=access_token_expires ) return Token(access_token=access_token, token_type=”bearer”) @app.get(“/users/me/”, response_model=User) async def read_users_me( current_user: Annotated[User, Depends(get_current_active_user)], ): return current_user @app.get(“/users/me/items/”) async def read_own_items( current_user: Annotated[User, Depends(get_current_active_user)], ): return [{“item_id”: “Foo”, “owner”: current_user.username}] {% endhighlight %} Middleware 미들웨어는 다양한 작동에 있어서 중간에 행동을 하는 행동입니다. 현재 fastapi는 http protocol이 진행되는 시점의 middleware만 사용이 가능합니다. 사용방법은 다음 두가지 방법을 활용할 수 있습니다. {% highlight python %} class CustomMiddleware(BaseHTTPMiddleware): def __init__(self, app): super().__init__(app) async def dispatch(self, request, call_next): response = await call_next(request) return response app.add_middleware(CustomMiddleware) {% endhighlight %} {% highlight python %} @app.middleware(“http”) async def add_process_time_header(request: Request, call_next): start_time = time.time() response = await call_next(request) process_time = time.time() - start_time response.headers[“X-Process-Time”] = str(process_time) return response {% endhighlight %} advance middleware CORS CORS는 Cross-Origin-Resource-sharing의 약자로 API를 호출하는 사용자의 ip를 관리하는 방법입니다. {% highlight python %} from fastapi.middleware.cors import CORSMiddleware origins = [ “http://localhost”, “http://localhost:8080”, ] app.add_middleware( CORSMiddleware, allow_origins=origins, allow_credentials=True, allow_methods=[“”], allow_headers=[“”], ) {% endhighlight %} Background Task main logic이 진행되는 것과 별개로 Background Task를 동작시킬 수 있습니다. 이렇게 동작되면 main logic에 영향을 주지 않는 동작을 빠르게 실행 할 수 있습니다. {% highlight python %} from fastapi import BackgroundTasks, FastAPI app = FastAPI() def write_log(message: str): with open(“log.txt”, mode=”a”) as log: log.write(message) @app.post(“/send-notification/{email}”) async def send_notification( email: str, background_tasks: BackgroundTasks ): message = f”message to {email}\n” background_tasks.add_task(write_log, message) return {“message”: “Message sent”} {% endhighlight %} Info fastapi는 기본적인 info를 지정 할 수 있습니다. 이는 docs를 통하여 보여지며 아래와 같이 작성이 가능합니다. {% highlight python %} app = FastAPI( title=”ChimichangApp”, description=description, summary=”Deadpool’s favorite app. Nuff said.”, version=”0.0.1”, terms_of_service=”http://example.com/terms/”, contact={ “name”: “Deadpoolio the Amazing”, “url”: “http://x-force.example.com/contact/”, “email”: “dp@x-force.example.com”, }, license_info={ “name”: “Apache 2.0”, “identifier”: “MIT”, }, ) {% endhighlight %} static files fastapi로 static file을 사용할 수 있습니다. 이러한 기능을 통하여 간단한 소개글이나 홈페이지 같은것도 제작이 가능해집니다. {% highlight python %} from fastapi import FastAPI from fastapi.staticfiles import StaticFiles app = FastAPI() app.mount(“/static”, StaticFiles(directory=”static”), name=”static”) {% endhighlight %} API Test fastapi를 작성하고나면 정상적으로 작성이 되었는지 걱정이 될 수 있습니다. 이럴때 test를 해볼 수 있는 기능을 지원합니다. 아래와 같이 작성 후 pytest를 통하여 확인을 해볼 수 있습니다. test_*.py {% highlight python %} from fastapi import FastAPI from fastapi.testclient import TestClient app = FastAPI() @app.get(“/”) async def read_main(): return {“msg”: “Hello World”} client = TestClient(app) def test_read_main(): response = client.get(“/”) assert response.status_code == 200 assert response.json() == {“msg”: “Hello World”} {% endhighlight %} run uvicorn(W. code) 다음과 같이 코드상에서 uvicorn을 작동하면 코드를 관리하는데 더욱 장점을 가질 수 있습니다. {% highlight python %} import uvicorn from fastapi import FastAPI app = FastAPI() @app.get(“/”) def root(): return {“value”: “hello world”} if name == “main”: uvicorn.run(app, host=”0.0.0.0”, port=8000) {% endhighlight %} Prev Prev
PYTHON
/
Framework
/
FastAPI
· 2024-02-05
FastAPI intro Num.2
Error handling API를 만드는데 있어서 Error 상황이 있을 수 있습니다. 이런경우 error 상태를 요청자에게 안내와 함께 시스템의 결과 관리를 해줘야 합니다. HTTPException 가장 일반적인 exception으로 아래와 같이 사용 가능합니다. 일반적으로 headers를 작성할 필요가 없으나 Token을 만들거나 하는 상황이나 보안적인 문제가 있을때 사용하기에 좋습니다. {% highlight python %} from fastapi import FastAPI, HTTPException app = FastAPI() items = {“foo”: “The Foo Wrestlers”} @app.get(“/items/{item_id}”) async def read_item(item_id: str): if item_id not in items: raise HTTPException( status_code=404, detail=”Item not found”, headers={“X-Error”: “There goes my error”}, ) return {“item”: items[item_id]} {% endhighlight %} Custom Exception Exception은 다음과 같이 custom해서 사용 가능합니다. {% highlight python %} 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} {% endhighlight %} 다음 보여드릴 방식은 기본 설정된 starlette, RequestValidation을 custom해서 추가 기능을 구현하려고 할때 활용 가능합니다. {% highlight python %} from fastapi import FastAPI from fastapi.exception_handlers import ( http_exception_handler, request_validation_exception_handler, ) from fastapi.exceptions import RequestValidationError from starlette.exceptions import HTTPException as StarletteHTTPException app = FastAPI() @app.exception_handler(StarletteHTTPException) async def custom_http_exception_handler(request, exc): print(f”OMG! An HTTP error!: {repr(exc)}”) return await http_exception_handler(request, exc) @app.exception_handler(RequestValidationError) async def validation_exception_handler(request, exc): print(f”OMG! The client sent invalid data!: {exc}”) return await request_validation_exception_handler(request, exc) {% endhighlight %} Routing routing을 하는 방식은 크게 2가지를 볼 수 있습니다. use tags {% highlight python %} from fastapi import FastAPI app = FastAPI() @app.get(“/items/”, tags=[“items”]) async def create_item(): return None {% endhighlight %} use router {% highlight python %} from fastapi import FastAPI, APIRouter app = FastAPI() router = APIRouter(prefix=’/slack’, tags=[“items”]) app.include_router(user.router) @router.get(“/items/”) async def create_item(): return None {% endhighlight %} JsonEncoder pydantic으로 인하여 dict형이 아니거나 2개이상의 dict형을 융합하여 활용하고 싶을때 사용할 수 있는 방법입니다. {% highlight python %} from fastapi.encoders import jsonable_encoder from pydantic import BaseModel class Item(BaseModel): title: str timestamp: datetime description: str | None = None @app.put(“/items/{id}”) def update_item(id: str, item: Item): jsonable_encoder(item) {% endhighlight %} Dependencies Dependencies는 FastAPI가 동작하는데 있어서 종속성을 주입하는 것입니다. 이는 더욱 안정적으로 동작할 수 있게 해줍니다. pydantic으로 사용하는것을 권장하나 아래와 같이 사용할 수 도 있습니다. Function Dependencies 다음과 같이 함수형으로도 만들 수 있습니다. {% highlight python %} from typing import Annotated from fastapi import Depends, FastAPI app = FastAPI() async def common_parameters( skip: int = 0, limit: int = 100 ): return {“skip”: skip, “limit”: limit} @app.get(“/items/”) async def read_items(commons: Annotated[dict, Depends(common_parameters)]): return commons {% endhighlight %} Class Dependencies 다음과 같이 클래스형으로 만들 수 있습니다. {% highlight python %} from typing import Annotated, Union from fastapi import Depends, FastAPI app = FastAPI() class CommonQueryParams: def init(self, skip: int = 0, limit: int = 100): self.skip = skip self.limit = limit @app.get(“/items/”) async def read_items(commons: Annotated[CommonQueryParams, Depends(CommonQueryParams)]): return commons {% endhighlight %} Path Operation Dependencies 다음과 같이 함수의 입력이 아니라 path operation dependency에 입력하여 값을 검증 할 수도 있습니다. {% highlight python %} from typing import Annotated from fastapi import Depends, FastAPI, Header, HTTPException app = FastAPI() async def verify_token(x_token: Annotated[str, Header()]): if x_token != “fake-super-secret-token”: raise HTTPException(status_code=400, detail=”X-Token header invalid”) @app.get(“/items/”, dependencies=[Depends(verify_token)]) async def read_items(): return None {% endhighlight %} Global Dependencies 전역 Dependency도 사용할 수 있습니다. 이는 token 검증과 같은 활동에 있어서 유용합니다. {% highlight python %} app = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)]) {% endhighlight %} Dependencies using yield Dependency를 사용하여 DB의 무결성도 확인이 가능합니다. 아래의 예시는 sqlalchemy를 사용하였으나 다음과 같이 구성하면 활용성을 향상시킬 수 있습니다. {% highlight python %} from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker engine = create_engine( , pool_pre_ping=True ) Session = sessionmaker(autocommit=False, autoflush=False, bind=engine) def get_db(): db = Session() try: yield db finally: db.close() def router(s: Annotated[Service, Depends(get_db)]): s.action() {% endhighlight %} {% highlight python %} class MySuperContextManager: def init(self): self.db = DBSession() def __enter__(self): return self.db def __exit__(self, exc_type, exc_value, traceback): self.db.close() async def get_db(): with MySuperContextManager() as db: yield db {% endhighlight %} 아래의 로직을 이해한다면 fastapi를 자유롭게 사용하는데 문제가 없다고 볼 수 있습니다. Prev Prev Next Next
PYTHON
/
Framework
/
FastAPI
· 2024-02-02
FastAPI intro Num.1
What is FastAPI FastAPI는 python으로 구현된 web framework입니다. FastAPI는 빠르게 개발이 가능하고 python으로 구현된 web framework중 가장 빠르다고 이야기 되고 있습니다. type check를 해주는 pydantic과 Asynchronous Server Gateway interface로 구현된 starlette가 구성되어 있습니다. 그에 따라 다음과 같은 장점을 가지게 됩니다. type check 비동기적 작업 지원 fast to create docs FastAPI는 docs를 자체적으로 지원을 해줘서 빠른 문서작업이 가능해집니다. FastAPI는 Swagger UI를 기반으로한 API docs를 지원해줍니다. 해당 자료는 FastAPI를 구동하면 https://localhost:port/docs에서 확인을 할 수 있습니다. FastAPI는 ReDoc을 기반으로한 API docs 또한 지원을 해줍니다. 해당 자료는 FastAPI 구동이후 https://localhost:port/redoc에서 확인을 할 수 있습니다. How to start FastAPI FastAPI를 사용하기 위하여 우선 두가지 모듈을 다운 받아야합니다. {% highlight shell %} pip install fastapi pip install uvicorn {% endhighlight %} 이후 실행을 위하여 아래의 코드 작성 및 실행이 필요합니다. main.py {% highlight python %} from fastapi import FastAPI app = FastAPI() @app.get(“/”) async def root(): return {“message”: “Hello World”} {% endhighlight %} you should run in cli {% highlight shell %} uvicorn main:app –reload {% endhighlight %} main:app은 main.py에 있는 app을 실행한다는 의미입니다. Input Datas API의 데이터 입출력을 위하여 다양한 방식의 input data가 존재합니다. 아래는 해당 값들을 순차적으로 설명드리겠습니다. Path Parameters 경로 파라미터라고 불리는 것으로 아래의 코드에서 item_id를 의미합니다. FastAPI는 type에 있어서 매우 엄격합니다. 그래서 아래의 코드와 같이 type hint(or pydantic)가 없거나 일치 하지 않을 경우 작동하지 않습니다. 주소창에 입력하는 값으로보면 숫자를 입력해도 문자로 보일 수 있으나 type hint에 따라서 숫자로 인식됩니다. {% highlight python %} @app.get(“/items/{item_id}”) async def read_item(item_id: int): return {“item_id”: item_id} {% endhighlight %} 위와 같은 주소체계를 사용할때 주의 할점은 "/items/item"이라고 만약 새로운 주소체계를 선언할 경우 path parameter로 구성된 주소체계보다 상단에 존재하여야합니다. 이는 중복된 입력으로 판단이 될 수 있기 때문입니다. Path Parameter가 "/"와 같이 경로 데이터가 포함이 되어있을 경우 starlette에서 지원해주는 type check 기능으로 :path를 추가해 아래와 같이 활용 가능합니다. {% highlight python %} @app.get(“/items/{item_id:path}”) async def read_item(item_id: str): return {“item_id”: item_id} {% endhighlight %} Path Parameter는 아래와 같이 상세한 유효성 검사를 지원하며 사용법은 다음과 같습니다. gt: 보다 크다. ge: 크거나 같다. lt: 보다 작다. le: 작거나 같다. {% highlight python %} from typing import Annotated from fastapi import FastAPI, Path app = FastAPI() @app.get(“/items/{item_id}”) async def read_items( q: str, item_id: Annotated[int, Path(title=”The ID of the item to get”)] ): results = {“item_id”: item_id} if q: results.update({“q”: q}) return results {% endhighlight %} Query Parameters Path Parameter이외에 입력값을 작성하면 Query Parameter로 인식됩니다. url로 값을 직접 전송하려면 URL/?data=data와 같이 전송이 가능합니다. Query Parameter의 경우 default값을 제공해주지 않거나 Ellipsis를 주면 필수 입력 조건으로 사용이됩니다. Query Parameter는 다음과 같이 유효성 검사를 추가적으로 진행 할 수 있습니다. 이것이 의미하는 바는 q는 str인데 None일 수 있으며 최대 글자수가 50이 넘으면 안되는 기본값 None을 가진 parameter라고 볼 수 있습니다. {% highlight python %} from typing import Annotated from fastapi import FastAPI, Query app = FastAPI() @app.get(“/items/”) async def read_items(q: Annotated[str | None, Query(max_length=50)] = None): results = {“items”: [{“item_id”: “Foo”}, {“item_id”: “Bar”}]} if q: results.update({“q”: q}) return results {% endhighlight %} Query에는 다음과 같이 다양한 입력을 줄 수 있습니다. alias: 입력 명칭을 변경 deprecated: 사용 가능 여부 include_in_schema: 문서에 표현 여부 {% highlight python %} Query( alias=”item-query”, title=”Query string”, description=”Query string for the items to search in the database that have a good match”, min_length=3, max_length=50, pattern=”^fixedquery$”, deprecated=True, include_in_schema=False ), {% endhighlight %} Body 다중 입력값이 존재할경우 흔하게 사용되는 방식입니다. fastapi에서 지원하는 Request를 사용해도 가능하지만 pydantic을 이용하는것이 무결성에 있어서 안전합니다. {% highlight python %} from fastapi import FastAPI from pydantic import BaseModel class Item(BaseModel): name: str description: str | None = None price: float tax: float | None = None app = FastAPI() @app.post(“/items/”) async def create_item(item: Item): return item {% endhighlight %} Single Body Body를 사용하는데 있어서 단일값일 경우 아래와 같이 사용이 가능합니다. {% highlight python %} from typing import Annotated from fastapi import Body, FastAPI app = FastAPI() @app.put(“/items”) async def update_item( importance: Annotated[int, Body()] ): return {“importance”: importance} {% endhighlight %} Body Field body도 query와 같이 Field를 사용하여 추가적인 무결성 검사를 할 수 있습니다. {% highlight python %} from typing import Annotated from fastapi import Body, FastAPI from pydantic import BaseModel, Field app = FastAPI() class Item(BaseModel): price: float = Field(gt=0, description=”The price must be greater than zero”) tax: float @app.put(“/items”) async def update_item(item: Annotated[Item, Body(embed=True)]): results = {“item_id”: item_id, “item”: item} return results {% endhighlight %} Cookie Cookie 데이터가 있을경우 아래와 같이 사용할 수 있습니다. {% highlight python %} from typing import Annotated, Union from fastapi import Cookie, FastAPI app = FastAPI() @app.get(“/items/”) async def read_items(ads_id: Annotated[Union[str, None], Cookie()] = None): return {“ads_id”: ads_id} {% endhighlight %} Header Header 데이터를 사용할 수 있습니다. Header의 경우 "-"이 "_"으로 자동 치환이 되기때문에 원본을 사용하려면 옵션을 추가해줘야 합니다. {% highlight python %} from typing import Annotated, Union from fastapi import FastAPI, Header app = FastAPI() @app.get(“/items/”) async def read_items(user_agent: Annotated[Union[str, None], Header(convert_underscores=False)] = None): return {“User-Agent”: user_agent} {% endhighlight %} Form Form data의 경우 우선 pip install python-multipart를 필요로 합니다. Form data는 아래와 같이 활용이 가능합니다. {% highlight python %} from typing import Annotated from fastapi import FastAPI, Form app = FastAPI() @app.post(“/login/”) async def login(username: Annotated[str, Form()], password: Annotated[str, Form()]): return {“username”: username} {% endhighlight %} File File data의 경우 우선 pip install python-multipart를 필요로 합니다. File data는 2가지 방식으로 활용 가능합니다. bytes: 이미지의 원본을 binary값으로 가져옵니다. UploadFile: 이미지의 관련 정보들을 가져옵니다. filename content_type file read write seek close {% highlight python %} from typing import Annotated from fastapi import FastAPI, File, UploadFile app = FastAPI() @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: Annotated[UploadFile, File()]): return {“filename”: file.filename} {% endhighlight %} Next Next
PYTHON
/
Framework
/
FastAPI
· 2024-02-01
<
>
Touch background to close
WTMO-dev
