Bkkhyunn

Bkkhyunn

배움과 경험을 기록하고,
세상과 사람을 매개하는
ML Engineer 가 되기까지

[FastAPI] FastAPI 확장 기능

앞선 포스트에서 FastAPI 활용을 위해 필요한 개념들을 정리했다. 이 포스트에서는 많이 사용하게 되는 FastAPI 확장 기능에 대해 정리해보자.

Pydantic

  • FastAPI 는 기본적으로 Pydantic 을 Data Model Class 로 사용하고 있다. Pydantic 은 Data Validation 및 Settings Management 를 용이하게 해주는 라이브러리다.
  • Pydantic 과 FastAPI 를 사용하여 아래와 같은 것들을 수행할 수 있다.
    • Type Hint 를 런타임에서 강제하여 안전하게 데이터 핸들링
    • Python 의 기본 데이터 타입인 str, int, list, dict, tuple 에 대한 Validation 지원
    • 기존 Validation 벤치마크 라이브러리 보다 빠른 속도
    • 머신러닝 feature data validation 으로 활용 가능하다. 예를 들어 Feature A 가 Int 타입이고 0 에서 100 사이의 값을 가져야 할 때, 데이터가 필요한 곳에 들어올 때 해당 정의에 위반되면 에러를 반환한다.
    from pydantic import BaseModel, Field, validator

class InputData(BaseModel):
    score: int

    @validator('score')
    def validate_score(cls, value):
        if value < 0 or value > 100:
            return ValueError('Score must be between 0 and 100')
        return value

Validation

  • 머신러닝 Model Input Validation 에 대해 좀 더 알아보자.
  • Online Serving 에서 input 데이터를 Validation 하는 case 를 보면, 원하지 않는 데이터를 제공받을 수도 있다. 즉 사용자가 이상한 데이터를 입력할 수도 있다. 이 때 쓰면 안되는 데이터들을 validation 으로 체크할 수 있다.
  • 어떤 웹 서비스에서 URL, 정수, 폴더 이름을 데이터로 받는다고 할 때, Validation Check Logic 은 아래와 같다.
    • 조건 1: 올바른 URL 을 입력 받았는지?
    • 조건 2: 1 에서 10 사이의 정수를 입력 받았는지?
    • 조건 3: 올바른 폴더 이름을 입력 받았는지?
  • 위와 같은 예시에서 Validation 할 때, 사용할 수 있는 방법들에 대해 알아보자.

Python Class

  • 일반적인 Python 클래스를 활용하는 방법이다. 클래스 내부에 validate logic 이 있다.

    class ModelInput:
    def __init__(self, url: str, rate: int, target_dir: str) -> None:
        self.url = url
        self.rate = rate
        self.target_dir = target_dir

    def _validate_url(self, url_like: str) -> bool:
        """
        올바른 URL 인지 검증

        Args:
            url_like (str): 검증할 URL

        Returns:
            bool: 검증 성공/실패 여부
        """
        from urllib.parse import urlparse

        try:
            result = urlparse(url_like)
            return all([result.scheme, result.netloc])
        except:
            return False

    def _validate_directory(self, directory_like: str) -> bool:
        """
        존재하는 디렉토리인지 검증

        Args:
            directory_like (str): 검증할 디렉토리 경로

        Returns:
            bool: 검증 성공/실패 여부
        """
        import os

        return os.path.isdir(directory_like)

    def validate(self) -> bool:
        """
        클래스 필드가 올바른지 검증

        Returns:
            bool: 검증/성공 실패 여부
        """
        validation_results = [
            self._validate_url(self.url),
            1 <= self.rate <= 10,
            self._validate_directory(self.target_dir),
        ]
        return all(validation_results)

  • 위와 같이 작성하게 되면 코드량이 의미없이 길어지게 되어 장황한 검증 로직이 된다. 위와 같은 validate logic 을 활용할 때는 아래와 같다.

    if __name__ == "__main__":
    import os

    VALID_INPUT = {
        "url": "https://content.presspage.com/uploads/2658/c800_logo-stackoverflow-square.jpg?98978",
        "rate": 4,
        "target_dir": os.path.join(os.getcwd(), "examples"),
    }

    INVALID_INPUT = {"url": "WRONG_URL", "rate": 11, "target_dir": "WRONG_DIR"}

    valid_python_class_model_input = ModelInput(**VALID_INPUT)
    assert valid_python_class_model_input.validate() is True

    invalid_python_class_model_input = ModelInput(**INVALID_INPUT)
    assert invalid_python_class_model_input.validate() is False
  • 이렇게 되면 input 정의 및 validation 로직 추가되어, 복잡함 검증 로직에는 Class method 가 복잡해지기 쉽다.
  • exception handling 을 커스텀하게 제어할 수 있지만, input 을 받아서 inference 를 수행하는 메인 로직에 집중하기 어려워진다.

Dataclass

  • 현업에서 많이 사용하는 방법으로, Python 3.7 이상이 필요하다.

  • 아래와 같이 dataclass 를 import 한 후 데코레이터를 사용한다.

    from dataclasses import dataclass
from pydantic.networks import HttpUrl
    
class ValidationError(Exception):
    pass

@dataclass
class ModelInput:
    url: str
    rate: int
    target_dir: str

    def _validate_url(self, url_like: str) -> bool:
        """
        올바른 URL 인지 검증

        Args:
            url_like (str): 검증할 URL

        Returns:
            bool: 검증 성공/실패 여부
        """
        from urllib.parse import urlparse
  
        try:
            result = urlparse(url_like)
            return all([result.scheme, result.netloc])
        except:
            return False

    def _validate_directory(self, directory_like: str) -> bool:
        """
        존재하는 디렉토리인지 검증

        Args:
            directory_like (str): 검증할 디렉토리 경로

        Returns:
            bool: 검증 성공/실패 여부
        """
        import os

        return os.path.isdir(directory_like)

    def validate(self) -> bool:
        """
        클래스 필드가 올바른지 검증
  
        Returns:
            bool: 검증/성공 실패 여부
        """
        validation_results = [
            self._validate_url(self.url),
            1 <= self.rate <= 10,
            self._validate_directory(self.target_dir),
        ]
        return all(validation_results)

    def __post_init__(self):
        if not self.validate():
            raise ValidationError("올바르지 않은 input 입니다.")
  • dataclass 데코레이터를 사용하기 때문에 __init__ method 를 따로 작성할 필요가 없다.
  • __post_init__ 매직 메서드를 사용하여 validate 를 수행하지만, 여전히 validation method 를 따로 만들어야 한다.

  • 아래와 같이 활용된다.

    if __name__ == "__main__":
    import os

    VALID_INPUT = {
        "url": "https://content.presspage.com/uploads/2658/c800_logo-stackoverflow-square.jpg?98978",
        "rate": 4,
        "target_dir": os.path.join(os.getcwd(), "examples"),
    }

    INVALID_INPUT = {"url": "WRONG_URL", "rate": 11, "target_dir": "WRONG_DIR"}

    valid_dataclass_model_input = ModelInput(**VALID_INPUT)
    assert valid_dataclass_model_input.validate() is True

    try:
        invalid_dataclass_model_input = ModelInput(**INVALID_INPUT)  # Error
    except ValidationError as exc:
        print("dataclass model input validation error", str(exc))
        pass
  • dataclass 를 사용하면 instance 생성 시점에서 validation 을 수행하기 쉽다. 이는 __post_init__ 메서드를 사용하기 때문에 validate 함수를 호출하지 않아도 생성 시점에 validation 이 수행되기 때문이다.
  • 코드 구현에 따라 에러가 발생하면 dataclass model input validation error 올바르지 않은 input 입니다. 라는 메시지가 나오게 된다.
  • 그러나 여전히 validate 로직들을 직접 작성해야 한다는 문제점이 있다. 그렇지 않으면 런타임에서 type checking 을 지원하지 않는다.

Pydantic

  • Pydantic 을 활용하면 import 만으로 해결할 수 있다.

    from pydantic import BaseModel, HttpUrl, Field, DirectoryPath

class ModelInput(BaseModel):
    url: HttpUrl
    rate: int = Field(ge=1, le=10)
    target_dir: DirectoryPath
  • 위 코드처럼 HttpUrl, Field, DirectoryPath 와 같은 Pydantic 에서 이미 정의해놓은 클래스를 사용할 수 있다.
  • 이를 통해 검증 로직이 매우 간소화되어 간결해진 코드로 validation 을 수행할 수 있다.
  • 또한 아래와 같은 장점이 있다.
    • HTTP URL, DB URL, enum 등 주로 쓰이는 타입들에 대한 Validation 이 이미 정의되어 있다.
    • 런타임에서 Type Hint 에 따라 Validation Error 를 발생시킨다.
    • Custom Type 에 대한 Validation 도 쉽게 사용할 수 있다. Custom Type 에는 Pydantic 공식문서를 참고하자.

  • 아래와 같이 활용할 수 있다.

    if __name__ == "__main__":
    import os

    VALID_INPUT = {
        "url": "https://content.presspage.com/uploads/2658/c800_logo-stackoverflow-square.jpg?98978",
        "rate": 4,
        "target_dir": os.path.join(os.getcwd(), "examples"),
    }

    INVALID_INPUT = {"url": "WRONG_URL", "rate": 11, "target_dir": "WRONG_DIR"}

    from pydantic import ValidationError

    valid_pydantic_model_input = ModelInput(**VALID_INPUT)
    try:
        invalid_pydantic_model_input = ModelInput(**INVALID_INPUT)  # error
    except ValidationError as exc:
        print("pydantic model input validation error: ", exc.json())
        pass
  • try except 구문을 활용하여 에러가 발생하면 exc.json() 으로 세부사항을 알 수 있다. 이는 앞선 코드들 보다 매우 간편하고, 어디에서 에러가 발생했는지 알 수 있다.

  • 아래와 같이 에러가 발생한 location(loc), 에러 발생 msg, type 등이 나온다.

    pydantic model input validation error:  [
  {
    "loc": [
      "url"
    ],
    "msg": "invalid or missing URL scheme",
    "type": "value_error.url.scheme"
  },
  {
    "loc": [
      "rate"
    ],
    "msg": "ensure this value is less than or equal to 10",
    "type": "value_error.number.not_le",
    "ctx": {
      "limit_value": 10
    }
  },
  {
    "loc": [
      "target_dir"
    ],
    "msg": "file or directory at path \"WRONG_DIR\" does not exist",
    "type": "value_error.path.not_exists",
    "ctx": {
      "path": "WRONG_DIR"
    }
  }
]
  • 이처럼 간결한 코드로 실행할 때마다 validation 을 확인할 수 있고, Error 에 대한 자세한 사항도 알 수 있기 때문에 Pydantic 은 머신러닝 Model Input 뿐 아니라 다른 Validation 에서도 매우 유용하게 사용할 수 있다.
  • 또한 Pydantic 에서는 자체적인 dataclass 를 제공한다. 앞선 포스트에서 정리했듯 중첩된(nested) BaseModel 은 Json Encoding 성능이 매우 안좋기 때문에 Response Model 에만 Pydantic Class 로 정의해주고 실제 반환에서는 ORJSONResponse 클래스를 사용해서 반환하는 것이 성능 상 유리하다.
  • BaseModel 은 내부적으로 Pydantic 의 데이터 검증/직렬화 기능을 제공하지만, 빠른 응답을 위한 ORJSONResponse 는 dictionary 나 JSON 직렬화 가능한 객체를 바로 요구한다.
  • BaseModel 을 쓰면 .dict() 를 써야 하므로 귀찮고 성능도 약간 손해를 볼 수 있지만, pydantic.dataclasses 의 @dataclass 는 Field() 와 함께 쓰면 검증도 되면서, ORJSONResponse 에서 바로 dict 처럼 쓸 수 있어 더 간결하고 빠르다.

Config

  • Config 는 앱을 기동하기 위해서 사용자가 설정해야 하는 일련의 정보다. 예를 들면 DB 에 관한 정보 등이 있다.
  • 일반적으로 이런 Config 들은 하나의 모듈이나 클래스로 관리해서 사용자가 보기 쉽게 저장한다.
  • Config 관리 시 사용할 수 있는 방법들에 대해 알아보자.

코드 내 상수로 관리

  • 로컬에서 굉장히 많이 쓰는 방법으로, 코드 내에서 상수로 관리한다.

    DB_URL = "mysql+pymysql://root:1234@localhost:3306/my_db"
DB_EHCO = True
  • 가장 간단하지만 보안 정보들이 코드에 그대로 노출되는 이슈가 발생한다.
  • 또한 일반적으로 개발 환경과 운영 환경을 분리하는데, 배포 환경에 따라 값을 다르게 줄 수 없다. 환경에 따라 DB 주소가 다른데, 배포할 때 코드는 변경할 수 없기 때문에 정보 변경이 어렵다.
  • 보안정보가 아닌 값들을 다루는 경우나, 테스트 같이 임시 환경에서만 사용하기 적합하다. 따라서 코드와 config 정의를 분리해서 관리하는 것이 필요하다.

yaml 등 파일로 관리

  • dev_config.yaml 과 같은 별도의 파일에서 관리 후, 실행할 때 파일을 지정해서 Config 클래스에 주입하는 방법이다.

    env: dev
db:
  username: user
  password: user
  host: localhost
  port: 3306
  database: dev

  • 이제 해당 config 를 사용하기 위해서 load_config 와 같은 함수가 필요하다.

    import os
from typing import Dict, Any
from yaml import load, FullLoader

def load_config(config_path: str) -> Dict[str, Any]:
    """
    config YAML 파일을 로드

    Args:
        config_path: config YAML 파일 경로
    Returns:
        Dict[str, Any]: Config dictionary
    """
    with open(config_path, "r") as f:
        config = load(f, FullLoader)
    return config

config = load_config(config_path="dev_config.yaml")

assert config["env"] == "dev"
expected = {"username": "user", "password": "user", "host": "localhost", "port": 3306, "database": "dev"}
assert config["db"] == expected
  • 배포 환경 별로 dev_config.yaml, prod_config.yaml 등을 다르게 생성할 수 있다. 그러나 보안 정보가 여전히 파일에 노출되므로, 배포 환경 별로 파일이 노출되지 않게 관리가 필요하다.
  • 이를 위해 yaml 파일을 쓰더라도 실제로 깃허브에 올리지 않고 클라우드 서비스에 주입이 될 수 있도록 설정하기도 한다.

  • 또한 flask 스타일로 Config 클래스에 주입하는 방법도 있다.

    import os
from typing import Dict, Any
import yaml

class Config(object):
    ENV: str = None
    TESTING: bool = False
    DB: Dict[str, Any] = {}

    @classmethod
    def from_yaml(cls, config_path: str):
        with open(config_path, "r") as config_file:
            config = yaml.load(config_file, Loader=yaml.FullLoader)

        cls.ENV = config["env"]
        cls.DB = config["db"]
        return cls

class DevConfig(Config):
    pass

class ProdConfig(Config):
    pass

config = DevConfig.from_yaml("dev_config.yaml")
assert config.ENV == "dev"
expected = {"username": "user", "password": "user", "host": "localhost", "port": 3306, "database": "dev"}
assert config.DB == expected

환경변수 및 Pydantic 으로 관리

  • 환경변수에 설정 값을 저장한 뒤, 코드에서 해당 환경변수를 읽어오는 방법이다.

  • pydantic_settings 의 BaseSettings 를 import 하고, DBConfig, AppConfig 등을 만든다. pydantic_settings 는 Pydantic 과는 별도로 설치해주어야 한다.

    import os
from pydantic import Field
from pydantic_settings import BaseSettings
from enum import Enum
from yaml import load, FullLoader

class ConfigEnv(str, Enum):
    DEV = "dev"
    PROD = "prod"

class DBConfig(BaseSettings):
    host: str = Field(default="localhost", env="db_host")
    port: int = Field(default=3306, env="db_port")
    username: str = Field(default="user", env="db_username")
    password: str = Field(default="user", env="db_password")
    database: str = Field(default="dev", env="db_database")

class AppConfig(BaseSettings):
    env: ConfigEnv = Field(default="dev", env="env")
    db: DBConfig = DBConfig()

with open("dev_config.yaml", "r") as f:
    config = load(f, FullLoader)

config_with_pydantic = AppConfig(**config)

assert config_with_pydantic.env == "dev"
expected = {"username": "user", "password": "user", "host": "localhost", "port": 3306, "database": "dev"}
assert config_with_pydantic.db.dict() == expected

# 환경변수로 필드를 오버라이딩
os.environ["ENV"] = "prod"
os.environ["DB_HOST"] = "mysql"
os.environ["DB_USERNAME"] = "admin"
os.environ["DB_PASSWORD"] = "SOME_SAFE_PASSWORD"

prod_config_with_pydantic = AppConfig()
assert prod_config_with_pydantic.env == "prod"
assert prod_config_with_pydantic.dict() != expected

# cleanup
os.remove("dev_config.yaml")
  • Validation 에서 본 것처럼, pydantic_settings.BaseSettings 을 상속한 클래스에서 Type Hint 로 주입된 설정 데이터를 검증할 수 있다.
  • Enum 은 Enumeration의 줄임말로, 미리 정의된 상수들의 집합을 만들 때 사용한다. 즉, 정해진 값들 중에서 하나만 선택할 수 있도록 강제하고, 코드의 명확성과 안전성을 높여주는 데 사용된다.
    • 위 ConfigEnv 클래스처럼 Pydantic 에서는 Enum 을 필드 타입으로 지정하면, 입력 값의 유효성이 자동으로 검사된다.
    • 따라서 AppConfig 의 env 필드는 ConfigEnv 타입으로 지정되어 있어기 때문에 "dev" 또는 "prod" 외의 값은 허용되지 않는다. 만약 "test" 같은 값을 넣으면 Pydantic 이 ValidationError 를 발생시킨다.
  • Field 클래스의 env 인자는 환경변수를 해당 필드로 오버라이딩 한다는 것을 의미한다.
  • 이를 통해 yaml, init 파일들을 추가적으로 만들지 않고, .env 파일들을 환경 별로 만들어두거나 실행 환경에서 유연하게 오버라이딩할 수 있다.
  • Github 에는 .env.example 와 같은 파일을 만들어서 올린다. 이 파일을 수정해서 환경변수로 사용하라는 뜻이다. 따라서 코드나 파일에 보안 정보가 노출되지 않는다.
  • 보통 환경변수는 배포할 때 주입하게 된다.
  • 총 3 가지 방법을 알아봤는데, 세 방법 모두 틀린 것은 아니다. 실무에서는 팀에서 지정한 방법을 따라가게 된다.
    • 협업하는 환경에서는 Human error 를 줄여주는 Pydantic 의 기능들이 유용하게 사용되고 있다.
    • FastAPI 를 쓴다면 pydantic_settings.BaseSettings 를 통해 config 를 관리하는 것을 권장한다.
    • 그러나 Pydantic 에는 속도 이슈가 있다. 이러한 장점과 단점을 알고 팀 단위로 취사 선택을 하는 것이 중요하다.

Dependency Injection

  • Dependency Injection 은 FastAPI 에서 재사용 가능한 컴포넌트를 효율적으로 관리하고 코드의 유지 보수를 용이하게 하는 데 사용된다. 이는 코드의 모듈성과 유연성을 향상시킨다.
  • Dependency Injection 은 두 가지로 분해해서 의미를 해석할 수 있다.
    • Dependency 는 개발자의 code 를 위해 사용하거나, 동작하기를 원하는 것을 선언하는 방식이다.
    • Injection 은 개발자의 code 에게 필요한 dependency 들을 제공해주는 어떤 것들을 관리하는 것을 의미한다.
    • 즉 이 둘을 합치면 개발을 하는데 있어 필요한 dependency 를 어떻게 system 이 injection 해주고, 관리해줄 것인가이다.
  • Dependency Injection 은 다음과 같은 장점이 있다.
    • 공통된 logic 을 재사용할 수 있다.
    • database connection 을 공유할 수 있다.
    • 보안적인 모듈들을 적용하기 쉽게 할 수 있다.
  • 의존성 주입의 유무 차이를 코드로 보자.
    • 아래는 의존성 주입이 고려되어 있지 않은 코드 구현이다.
    class User:
    def __init__(self):
        self.remocon = SamsungTVRemocon()
    def volume_up(self):
        self.remocon.volume_up()
    def volume_down(self):
        self.remocon.volume_down()
    • 해당 예시는 User 클래스가 SamsungTVRemocon 객체에 의존하고 있다. 즉 다른 종류의 리모컨을 사용하고 싶다면 User 클래스의 코드를 직접 수정해야 할 수도 있다. 이는 유연성이 떨어지고 코드 변경에 취약한 상태를 만들 수 있다.
    • 반면 아래의 코드를 보자.
    class User:
    def __init__(self, remocon):
        self.remocon = remocon
    def volume_up(self):
        self.remocon.volume_up()
    def volume_down(self):
        self.remocon.volume_down()
    • 해당 코드는 외부에서 리모콘 Class 객체를 주입받아 사용하는 방식이다. 이 경우 User 클래스가 어떤 종류의 리모컨을 사용하는지에 대해 알 필요가 없으며, 어떤 종류의 리모컨도 사용할 수 있다.
    • 이것이 바로 dependency injection 의 핵심이다. User 클래스는 자신이 필요로 하는 의존성을 외부에서 주입받아 사용하므로, 코드 변경이 쉽고 유연하며 재사용성이 높아진다.
    • 이처럼 의존성을 외부에서 주입하는 방식으로 설계하면 객체 간의 결합도를 낮추면서도 객체의 응집력을 높일 수 있어 좋은 설계로 이어질 수 있다.
  • 그럼 FastAPI 에서 dependency injection 을 어떻게 제공하고 있는지 알아보자.
    • 위 예제에서의 객체와 객체 간의 개념과 비슷하게 공통 기능들을 별도 함수로 정의하고 router 함수에 주입하여 사용한다.
    • 일반적으로 API Key 인증, 값 검사와 같이 공통으로 필요한 연산들을 의존성 함수로 작성한다.
  • Depends() 라는 함수를 이용해서 의존성을 주입할 수 있다. 이전 포스트에서 정리한 Form() 과 사용법이 같다.
  • Depends() 안에는 하나의 함수가 들어갈 수 있는데, 이 함수는 route handler 에서 받는 parameter 들과 동일한 parameter 들을 받을 수 있다.

  • route handler 는 server 에 전달된 request 를 처리하는 Endpoint 함수와 의미가 같다. 예를 들면 아래의 read_items 와 같은 함수다.

    from typing import Annotated, Union
from fastapi import Depends, FastAPI

app = FastAPI()

def common_parameters(
    q: Union[str, None] = None, skip: int = 0, limit: int = 100
    ):
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons

@app.get("/users/")
def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons
  • 위처럼 Depends() 를 route handler 에 사용하되 인자로 하나의 함수를 전달해준다. 이 함수 안에서 개발자가 원하는 처리가 이루어지고, 그 결과를 Dependency 로 handler 에 전달하는 것이다. 위에서는 commons 가 dependency 가 된다.
  • 그렇다면 Depends() 에 들어가는 함수는 Query parameter 와 같은 HTTP request 를 통한 입력값들을 받고, 이를 선제적으로 처리하여 json 으로 반환하는 것이 전부다.
  • 즉, 함수 common_parameters 는 handler 와 크게 다르지 않으며 @app.get() 과 같은 path operation decorator 만 없는 형식으로 볼 수 있다.
  • 위 예제에서는 commons 의 type 을 Annotated 를 사용하여 dict 로 잡아놓았기 때문에 editor 에서 자동완성이 불가능하다.
  • common_parameters 는 dependency 로서 하나의 함수로 적혀있지만, dependency 의 조건은 함수라기 보다는 callable 이다. 즉 __call__ 을 구현한 모든 것들은 dependency 로 쓰일 수 있는 것이다.
  • 따라서 Class 도 하나의 dependencies 로 동작할 수 있다.

  • dependency 로 하나의 함수를 썼을 때는 FastAPI 에서 들어온 parameter 를 자동으로 할당해준다. Class 역시 마찬가지로 인스턴스를 생성하는 __init__ 에 해당 parameter 들이 들어간다.

    from typing import Annotated, Union
from fastapi import Depends, FastAPI

app = FastAPI()

fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]

class CommonQueryParams:
    def __init__(self, q: Union[str, None] = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
def read_items(commons: Annotated[CommonQueryParams, Depends(CommonQueryParams)]):
    response = {}
    if commons.q:
        response.update({"q": commons.q})
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
    response.update({"items": items})
    return response
  • 이를 이용하면 여러 Depends() 를 연결하여 하나의 의존성 pipeline 을 만들 수도 있다. 또한 동일한 요청 내에서 동일한 함수를 Depends 로 호출하는 경우 캐시된 값을 재사용할 수도 있다.
  • 이러한 dependency 는 위 예시들처럼 Router 함수의 인자 기본값 이외에 FastAPI 객체, APIRouter 객체, Router 데코레이터에도 넣을 수 있다.
    • FastAPI, APIRouter 에 추가하는 경우 포함된 모든 Endpoint 에 적용이 된다. 주로 인증, 값 검증과 같이 검사할 때 여기에 붙인다.
    • Router 데코레이터, Router 함수에 추가하는 경우 해당 Endpoint 에만 적용이 된다. Router 데코레이터에 붙이는 경우에도 FastAPI, APIRouter 에 추가하는 경우와 동일하지만, Router 함수에 추가하는 경우는 현재 요청을 보낸 유저의 정보와 같이 어떤 값을 주입하기 위해 사용한다.
  • 이 블로그 글에 더 심화/응용된 내용들이 잘 정리되어 있다. 참고해보자.

FastAPI 확장 기능

  • 앞서 살펴본 Pydantic 말고도 FastAPI 에는 매우 다양한 확장 기능이 있다. 모두 알면 좋지만, 이 섹션에서는 자주 사용되는 기능들을 정리해보자.

Lifespan Function

  • FastAPI 앱을 실행할 때와 종료할 때, 로직을 넣고 싶은 경우에 사용하는 기능이다. Lifespan 은 라이프사이클을 의미한다.

  • 예를 들어 FastAPI 앱이 처음 실행될 때 머신러닝 모델을 Load 하고 앱을 종료할 때 연결해 둔 Database Connection 을 정리하도록 할 수 있다.

    from contextlib import asynccontextmanager
from fastapi import FastAPI
import uvicorn
  
items = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    print("Start Up Event")
    items["foo"] = {"name": "Fighters"}
    items["bar"] = {"name": "Tenders"}

    yield

    print("Shutdown Event!")
    with open("log.txt", mode="a") as log:
        log.write("Application shutdown")

app = FastAPI(lifespan=lifespan)

@app.get("/items/{item_id}")
def read_items(item_id: str):
    return items[item_id]

if __name__ == '__main__':
    uvicorn.run(app, host="0.0.0.0", port=8000)
  • @asynccontextmanager 를 데코레이터로 쓰고, 실행하고자 하는 로직을 비동기인 async def 함수로 정의한다.
  • 그리고 FastAPI 객체를 생성할 때 lifespan 인자로 해당 함수(lifespan)를 넣어준다.
  • yield 가 앱이 가동되는 시점을 의미한다. 즉 yield 를 기점으로 이전에는 앱 시작 전에 실행해야 할 것들, 이후에는 앱 종료 전에 실행될 것들을 정의하는 것이다.

  • 그렇게 되면 아래와 같이 앱 실행 시 print 되도록 만든 Start Up Event 이 출력되고, 앱 종료시 print 되도록 만든 Shutdown Event! 가 출력된다.

    Untitled

  • Lifespan Function 기능을 이용해서 앱이 가동될 때 만들어 둔 ML 모델을 load 하도록 한다.

  • 또한 앱이 꺼질 때 로그 기록, 모델 리소스 수거, 데이터 처리 등 수행하고자 하는 것을 lifespan 함수에 구현해서 사용하면 된다.

    from contextlib import asynccontextmanager
from fastapi import FastAPI

def fake_answer_to_everything_ml_model(x: float):
    return x * 42

ml_models = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Load ML model
    ml_models["answer_to_everything"] = fake_answer_to_everything_ml_model

    yield

    # Clean up the ML model and release the resources
    ml_models.clear()

app = FastAPI(lifespan=lifespan)

@app.get("/predict")
async def predict(x: float):
    result = ml_models["answer_to_everything"](x)
    return {"result" : result}

API Router

  • 개발하면서 API 엔드포인트가 점점 많아지면, @app.get, @app.post 와 같은 path operation decorator 를 하나의 모듈에서 관리하기가 어려워질 수 있다.
  • 점점 많아지는 API 엔드포인트를 좀 더 효율적으로 관리하기 위해서 API Router 를 사용할 수 있다.
  • API Router 는 큰 애플리케이션에서 많이 사용되며 API Endpoint 를 정의한 Mini FastAPI 라고 생각할 수 있다. 이렇게 만들어진 여러 API Endpoint 를 연결해서 활용한다.

  • 아래 코드 예시처럼 기존에 사용하던 @app.get, @app.post 등의 path operation decorator 를 사용하지 않고, APIRouter 를 import 하여 router 를 별도로 정의하고 해당 router 에 엔드포인트 함수를 정의한다.

    from fastapi import FastAPI, APIRouter
import uvicorn

user_router = APIRouter(prefix="/users")
order_router = APIRouter(prefix="/orders")

@user_router.get("/", tags=["users"])
def read_users():
    return [{"username": "Rick"}, {"username": "Morty"}]

@user_router.get("/me", tags=["users"])
def read_user_me():
    return {"username": "fakecurrentuser"}

@user_router.get("/{username}", tags=["users"])
def read_user(username: str):
    return {"username": username}

@order_router.get("/", tags=["orders"])
def read_orders():
    return [{"order": "Taco"}, {"order": "Burritto"}]

@order_router.get("/me", tags=["orders"])
def read_order_me():
    return {"my_order": "taco"}

@order_router.get("/{order_id}", tags=["orders"])
def read_order_id(order_id: str):
    return {"order_id": order_id}

app = FastAPI()

if __name__ == '__main__':
    app.include_router(user_router)
    app.include_router(order_router)
    uvicorn.run(app, host="0.0.0.0", port=8000)
  • 실제 API router 를 쓸 때는 필요한 Endpoint 마다 user.py 나 order.py 처럼 파일을 따로 만들고 import 해서 사용한다. 실제로 현업에서는 각각의 router 는 도메인 혹은 그룹 단위로 파일을 분리하여 관리한다.
  • 위 코드를 보면 APIRouter 를 이용해서 user Router, order Router 2개를 생성한다. prefix 는 Endpoint 를 뜻한다.
  • 간단한 형태로는 app 에 FastAPI 객체를 만들어서 주입했다. 여기서는 user_router, order_router 를 path operation decorator 로 사용한다.
  • 이후 app.include_router 로 해당 router 들을 app 에 연결시킨다. 실제로 api 폴더에 Endpoint 별 파일을 따로 만든다면, 아래와 같이 연결시킬 수 있다.
    • include_router 는 FastAPI 객체에 Router 를 포함시키는 메소드다.
    import uvicorn
from api import user, order
from fastapi import FastAPI

app = FastAPI()

app.include_router(user.router)
app.include_router(order.router)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)
  • @user_router.get("/", tags=['users']) 는 루트(/) 에 접근했을 때 read_users 함수가 실행된다고 볼 수 있지만, user_router 의 prefix 가 /users 이다.
  • 따라서 여기서 GET 으로 접근한 루트는 /users 다. 이는 order_router 도 같은 방식으로 동작한다.
  • 그러면 read_user_me 함수는 /users/me 에 접근해야 실행된다.
  • @user_router.get("/{username}", tags=["users"]) 에서는 username 을 Path parameter 를 통해 변수로 받아서 read_user 함수를 실행한다. 이는 /users/username 안에서 리턴된다.

  • 이를 실행 해보면 아래와 같은 결과를 얻을 수 있다.

    Untitled

    Untitled

  • 웹 애플리케이션을 구현하다 보면 이러한 router 들이 많아진다.

Project Structure

  • 코드들이 점점 많아짐에 따라, 프로젝트 구조를 가독성 좋게 잡는 것이 중요하다. 이는 해당 포스트에서 정리한 바 있다.

  • 여기서는 아래와 같은 또 다른 예시의 프로젝트 구조를 보자.

    ml_app/                            # 프로젝트 루트
├── app/                           # 메인 app
│   ├── __init__.py                # App, Python Package 생성을 위한 파일
│   ├── main.py                    # main module, import app.main
│   ├── dependencies.py            # 의존성 Module, import app.dependencies
│   ├── routers/                   # Sub FastAPI (Python Subpackage)
│   │   ├── __init__.py            # Python Subpackage routers 생성
│   │   ├── items.py               # items Submodule, import app.routers.items
│   │   └── users.py               # users Submodule, import app.routers.users
│   └── internal/                  # internal Python Subpackage
│       ├── __init__.py
│       └── admin.py
├── tests/                         # 테스트 코드
│   ├── __init__.py
│   └── test_endpoints.py          # API 엔드포인트 테스트
├── .env                           # 환경 변수 파일
├── pyproject.toml                 # Poetry 의존성 및 메타 설정
├── poetry.lock                    # 패키지 버전 잠금 파일 (자동 생성)
├── README.md                      # 프로젝트 설명 문서
└── scripts/                       # 개발/운영용 스크립트 (선택사항)
    └── train_model.py             # ML 모델 학습 및 저장 스크립트
  • 이전 포스트에서 본 프로젝트 구조나, 위에서 본 프로젝트 구조나 정답은 없다.
  • 회사마다 굉장히 다양하게 구조를 잡는다. 실제 현업에서는 모델, DB, 클래스 객체가 많아서 복잡할 수 있다. 이렇게 복잡한 구조는 디자인 패턴과 관련되어 있다.

Project Templates

  • 많은 사람들이 프로젝트 구조에 대한 고민이 많아 템플릿을 서로 공유한다. 쿠키를 만들 때 사용하는 Cookiecutter 처럼, 미리 만들어진 틀을 공유하는 것이다.
  • https://github.com/cookiecutter/cookiecutter 에서 다양한 언어를 기반으로 하는 Template 들을 확인할 수 있다.
  • FastAPI 에 대한 Coookiecutter 도 만들어진 것이 있다. https://github.com/arthurhenrique/cookiecutter-fastapi 를 확인해보자.
  • Cookiecutter 를 설치하면, CLI 형태로 프로젝트를 생성할 때 용이하다.
  • 개인용 Cookiecutter 템플릿을 만드는 것도 좋은 방법이다. 회사에서 많이 사용하거나, 공통의 프로젝트 구조가 필요하다면 Cookiecutter 로 만드는 것도 좋다.

Error Handler

  • 서버에서 요청을 처리하다가 발생하는 각종 에러는 어떻게 처리할까? Error Handling 기능은 웹 서버를 안정적으로 운영하기 위해 반드시 필요한 이슈다.
  • 서버에서 Error 가 발생한 경우, 먼저 어떤 Error 가 발생했는지 알아야 한다. 그리고 요청한 클라이언트에 해당 정보를 전달해 대응할 수 있어야 한다. 실제로 어떤 웹 사이트든 흔하게 오류가 발생했으니 기다려달라는 렌더링을 볼 수 있다.
  • Error 처리의 중요성은 아래와 같다.
    • 백엔드 서버에서의 에러 처리는 시스템의 안정성과 신뢰성을 높이는 데 핵심적인 역할을 한다. 먼저, 사용자 경험을 향상시키기 위해 사용자가 이해할 수 있는 명확한 에러 메시지를 제공하여 사용자가 문제를 쉽게 해결할 수 있도록 돕는다.
    • 시스템의 안정성을 보장하기 위해 예상되는 에러를 사전에 처리하고, 예기치 않은 상황에 대비하여 시스템이 예상치 못한 상황에서도 안정적으로 동작할 수 있도록 한다.
    • 백엔드와 프론트엔드 간의 협업에서도 에러 처리는 중요하다. 프론트엔드와 백엔드 간의 통신에서 발생하는 에러를 명확하게 정의하고 적절하게 처리함으로써 사용자 경험을 개선하고 효율적인 개발을 가능하게 한다.
    • 보안 취약점을 방지하기 위해 에러 처리는 필수적이다. 따라서 프론트엔드와 백엔드 간에는 상호 간의 에러 처리를 공유하고 의사소통하여 시스템의 안정성과 보안을 유지하고 향상시키는 것이 중요하다.
  • 서버에서 에러에 대한 결과를 내려주고 클라이언트가 에러 포인트를 받아서 어떤 메시지를 출력하도록 하는 것이 에러에 대한 디자인(Error Handler)이다.
  • 서버 개발자는 모니터링 도구를 사용해 Error log 를 수집해야 한다. 그리고 발생하고 있는 오류를 빠르게 수정할 수 있도록 예외 처리를 잘 만들 필요가 있다.
  • 에러에 대한 로그의 경우, 서버 로컬이나 클라우드 같은 외부 저장 등을 고려할 수 있다. Monitoring Tool 중 하나인 Prometheus등에 로그를 보내서 로그 분석을 하기도 한다.

  • Error Handler 가 없는 경우에 어떻게 되는지 보자. 먼저 아래와 같이 코드를 짜보자.

    from fastapi import FastAPI
import uvicorn

app = FastAPI()

items = {
    1: "Boostcamp",
    2: "AI",
    3: "Tech"
}

@app.get("/v1/{item_id}")
async def find_by_id(item_id: int):
    return items[item_id]

if __name__ == '__main__':
    uvicorn.run(app, host="0.0.0.0", port=8000)
  • 위 코드에서는 item_id 가 4 이상인 값이 되면 KeyError 가 발생한다.

  • 이를 Swagger 로 확인해보면, 먼저 item_id 가 1 일 경우 정상적으로 200 Status Code 와 함께 적절한 값을 반환한다.

    Untitled

  • 그러나 item_id 에 5 가 들어갈 경우, 아래 그림과 같이 500 Status Code 와 함께 Internal Server Error 가 발생한다. 이러한 오류에 대응하기 위해 오류 처리 로직을 구현해야 한다.

    Untitled

  • Internal Server Error 500 이 Return 되면 클라이언트에서는 어떤 에러가 난 것인지 정보를 얻을 수 없고, 자세한 에러를 보려면 서버에 직접 접근해서 로그를 확인해야 한다.
  • Error Handling 을 더 잘 하려면 에러 메시지와 에러의 이유 등을 클라이언트에 전달하도록 코드를 작성해야 한다.

  • FastAPI 의 HTTPException 은 Error Response 를 더 쉽게 보낼 수 있도록 하는 Class 다. 이를 try except 구문으로 감싸서 만든다.

    from fastapi import FastAPI, HTTPException
import uvicorn

app = FastAPI()

items = {
    1: "Boostcamp",
    2: "AI",
    3: "Tech"
}

@app.get("/v2/{item_id}")
async def find_by_id(item_id: int):
    try:
        item = items[item_id]
    except KeyError:
        raise HTTPException(status_code=404, detail=f"아이템을 찾을 수 없습니다 [id: {item_id}]")
    return item

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

  • 이를 통해 아래 그림과 같이 좀 더 자세한 에러 메시지를 보내줄 수 있기 때문에, 클라이언트는 어떤 에러 이슈가 있는지를 알 수 있다.

    Untitled

  • 이런 방식으로 Error Handling 을 통해 클라이언트에게 어떤 에러가 발생했는지를 알려주도록 구현해주면 좋다.
  • 일반적으로 현업에서는 서비스 내에서 사용할 Error 케이스별로 Error 클래스를 정의한다. 이후 특정 Error 들에 대해 처리할 로직을 작성한다.
  • 이 때 Error Handler 는 Router 함수 Scope 안쪽에서만 작동한다. 만약 Middleware 와 같이 Router 함수 바깥의 Scope 에서 Error 를 Raise 시키면 Handler 를 통한 예외처리가 이루어지지 않을 수 있으니 주의해야 한다.
    • 간단히 말하면 Router 함수 내에서 Raise 한 Error 에 대해서만 Error Handler 가 처리할 수 있다.
    • 바깥 Scope 에 대해서는 직접 Error 를 담은 Response 를 반환해야 한다.

Background Tasks

  • 반환 시간이 오래 걸리는 API 엔드포인트는 클라이언트에게 데이터를 받았을 때 비동기(asynchronous)로 처리하도록 하는 것이 좋다.
  • 사용자 입장에서 웹 서비스의 대기시간이 길수록 이탈할 가능성이 높기 때문이다.
  • 작업을 비동기로 처리하기 위해서 Background Task 기능을 이용할 수 있다. 이는 오래 걸리는 작업들을 background 에서 실행하도록 한다.
  • Online Serving 에서 CPU 사용이 많은 작업들을 Background Task 로 사용하면, 클라이언트는 작업 완료를 기다리지 않고 다른 작업을 하면서 Response 를 받아볼 수 있다. 예를 들어 특정 작업이 완료되면 이메일을 전송하는 방식이 그렇다.

  • Background Tasks 를 사용하지 않는 작업들은 작업이 완료되는 시간 만큼 응답을 기다려야 한다.

    import contextlib, josin, threading, time
from datetime import datetime
from time import sleep
from typing import List

import requests
import uvicorn
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel, Field

class Server(uvicorn.Server):
    def install_signal_handlers(self):
        pass

    @contextlib.contextmanager
    def run_in_thread(self):
        thread = threading.Thread(target=self.run)
        thread.start()
        try:
            while not self.started:
                time.sleep(1e-3)
            yield
        finally:
            self.should_exit = True
            thread.join()

def run_tasks_in_fastapi(app: FastAPI, tasks: List):
    """
    FastAPI Client 를 실행하고, task 를 요청

    Returns:
        List: responses
    """
    config = uvicorn.Config(app, host="127.0.0.1", port=5000, log_level="error")
    server = Server(config=config)
    with server.run_in_thread():
        responses = []
        for task in tasks:
            response = requests.post("http://127.0.0.1:5000/task", data=json.dumps(task))
            if not response.ok:
                continue
            responses.append(response.json())
    return responses

app_1 = FastAPI()

def cpu_bound_task(wait_time: int):
    sleep(wait_time)
    return f"task done after {wait_time}"

class TaskInput(BaseModel):
    wait_time: int = Field(default=1, le=10, ge=1)

@app_1.post("/task")
def create_task(task_input: TaskInput):
    return cpu_bound_task(task_input.wait_time)

tasks = [{"wait_time": i} for i in range(1, 10)]

start_time = datetime.now()
run_tasks_in_fastapi(app_1, tasks)
end_time = datetime.now()
print(f"Simple Tasks: Took {(end_time - start_time).seconds}")
  • 위 코드를 통해 FastAPI 로 띄운 웹 서버에 POST 요청을 통해 작업을 실행하면 작업 시간 만큼 응답을 기다려야 한다.
  • 즉 run_tasks_in_fastapi 을 통해 /task 로 POST 를 요청할 시 task 마다 wait_time 을 계속 기다려야 한다.
  • Background Tasks 를 사용한 작업들은 기다리지 않고 바로 응답을 주기 때문에 0 초가 소요된다.

  • 아래 코드는 FastAPI 의 BackgroundTasks 를 이용하여 실제 작업이 Background 에서 실행된다.

    import contextlib, josin, threading, time
from datetime import datetime
from time import sleep
from typing import List

import requests
import uvicorn
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel, Field

app_2 = FastAPI()

# 비동기 작업이 등록됐을 때, HTTP Response 202 (Accepted)를 보통 리턴한다. https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/202
@app_2.post("/task", status_code=202)
async def create_task_in_background(task_input: TaskInput, background_tasks: BackgroundTasks):
    background_tasks.add_task(cpu_bound_task, task_input.wait_time)
    return "ok"

start_time = datetime.now()
run_tasks_in_fastapi(app_2, tasks)
end_time = datetime.now()
print(f"Background Tasks: Took {(end_time - start_time).seconds}")
  • BackgroundTasks 를 import 하고, Background Tasks 를 실행할 엔드포인트에서 background_tasks.add_task 를 통해 작업을 추가하면 된다. 여기에는 background 에서 실행할 함수와 해당 함수의 인자를 넣어준다.

  • 작업 결과물을 조회할 때는 Task 를 어딘가에 저장해두고, GET 요청을 통해 Task 가 완료됐는지 확인하도록 할 수 있다.

    import contextlib, josin, threading, time
from datetime import datetime
from time import sleep
from typing import List

import requests
import uvicorn
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel, Field
from uuid import UUID, uuid4

app_3 = FastAPI()

class TaskInput2(BaseModel):
    id_: UUID = Field(default_factory=uuid4)
    wait_time: int

task_repo = {}

def cpu_bound_task_2(id_: UUID, wait_time: int):
    sleep(wait_time)
    result = f"task done after {wait_time}"
    task_repo[id_] = result

@app_3.post("/task", status_code=202)
async def create_task_in_background_2(task_input: TaskInput2, background_tasks: BackgroundTasks):
    background_tasks.add_task(cpu_bound_task_2, id_=task_input.id_, wait_time=task_input.wait_time)
    return task_input.id_

@app_3.get("/task/{task_id}")
def get_task_result(task_id: UUID):
    try:
        return task_repo[task_id]
    except KeyError:
        return None
  • 이렇게 하면 task 를 background 로 보내고, 해당 task 의 완료 여부를 확인할 수 있다.
  • 참고로 uuid 는 네트워크 상에서 고유성이 보장되는 id를 만들기 위한 표준 규약이다. 128 비트의 숫자이며, 32 자리의 16 진수로 표현한다.
  • 여기에 550e8400-e29b-41d4-a716-446655440000 와 같이 8-4-4-4-12 글자마다 하이픈을 집어넣어 5 개의 그룹으로 구분한다.

Middleware

  • FastAPI 에서 클라이언트의 요청(Request)을 받고, 응답(Response)을 반환하기 전/후에 실행되는 중간 처리 계층이다.
  • FastAPI 공식문서에서는 Middleware 를 특정 경로 연산에 의해 처리되기 전에 모든 요청에 대해 작동하는 기능이며, 또한 응답을 반환하기 전의 모든 응답에서도 작동한다고 설명한다.
  • 쉽게 말하면, Router 에 작성하는 Endpoint 함수 앞뒤로 공통 로직을 넣고 싶은 경우에 사용한다.
  • 요청/응답 로깅, 사용자 인증 처리(토큰 검사), 요청 제한, CORS, GZip 압축, 요청 시간 측정, 트래픽 제한 등의 공통 기능을 구현할 때 유용하다.

  • 아래는 요청 처리 시간을 로깅하는 함수 형태의 Middleware 예시다.

    from fastapi import FastAPI, Request
import time

app = FastAPI()

@app.middleware("http")  # HTTP 요청 전후에 실행될 미들웨어를 선언
async def log_request_time(request: Request, call_next):
    """
    Args:
        request (Request): 들어오는 HTTP 요청 객체
        call_next: 요청을 다음 처리 단계(예: 라우터 함수)로 넘김
    """
    start_time = time.time()

    # 요청을 다음 단계로 넘김 (예: 라우터 → 응답 생성)
    response = await call_next(request)   # Endpoint 함수에서 응답을 생성하고 다시 받아옴

    # 응답 처리 후 시간 측정
    process_time = time.time() - start_time   # 요청부터 응답까지 걸린 처리 시간 측정
    print(f"Request: {request.url.path} completed in {process_time:.4f}s")

    return response

@app.get("/hello")
async def hello():
    return {"message": "Hello World"}
  • 이외에 starlette BaseHTTPMiddleware 클래스를 상속받아 클래스 형태의 Middleware 를 정의할 수 있다. FastAPI 는 내부적으로 starlette 를 사용하고 있기 때문에 서로 호환된다.

  • 아래와 같이 async def dispatch 에 기존 함수 형태의 Middleware 와 동일하게 작성하면 된다.

    from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware

class CustomHeaderMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        response = await call_next(request)
        response.headers['Custom'] = 'Example'
        return response

app = FastAPI()
app.add_middleware(CustomHeaderMiddleware)

  • FastAPI 는 일부 기본 Middleware 들을 제공하고 있다. 대표적으로 CORS 가 있다.

    from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],     # 허용할 origin
    allow_credentials=True,
    allow_methods=["*"],     # 허용할 HTTP 메소드
    allow_headers=["*"],     # 허용할 헤더
)
  • CORS(Cross-Origin Resource Sharing)는 웹 브라우저에서 다른 출처(origin)의 리소스에 접근할 수 있도록 제한하거나 허용하는 보안 메커니즘이다.
    • 브라우저는 기본적으로 다른 출처(origin)의 요청을 차단한다. 이는 보안 때문으로, 예를 들어 악성 사이트가 사용자의 쿠키를 이용해서 다른 사이트에 요청을 보내는 것을 막기 위해서다.
    • 쉽게 말해 브라우저에서 클라이언트 자바스크립트 코드로 요청하는 것을 막는 것으로, 브라우저를 막는다고 생각하면 쉽다.
    • 그렇다면 CORS 는 허가된 곳에서 요청된 API call 이 아니면 처리해주지 않는 것이라 이해할 수 있다.
  • 주의할 점은 서버는 실제로 응답을 보내고 있음에도, 브라우저가 차단하는 것이기 때문에 백엔드 로그에는 아무 문제가 없어 보일 수 있다.
  • curl, Postman, Python requests 등의 서버 간 통신은 브라우저가 아니므로 CORS 의 영향을 받지 않는다.
  • 만약 프로젝트가 커져 웹 서버(React, Next.js 등)와 WAS(FastAPI), DB(MySQL), ML 서버를 각각 두고 포트가 다 다르다고 가정해보자.
  • 이 때 브라우저 기반 클라이언트에서 HTTP 요청을 보낸다면 브라우저는 서버 응답을 받더라도, 보안 위반(CORS 오류)으로 응답을 JS 코드에 전달하지 않는다.

  • 개발자 도구 콘솔에서는 아래와 같은 에러를 볼 수 있다.

    Access to fetch at 'http://localhost:8000/predict' from origin 'http://localhost:3000' has been blocked by CORS policy...
  • 따라서 API 서버에 정의된 모든 Endpoint 에 대해 CORS 를 적용하기 위해서는 CORS 검사 로직을 Middleware 로 FastAPI 에 추가할 수 있다. 위 코드와 같이 FastAPI 가 제공하는 CORSMiddleware 를 사용하여, 백엔드에서 CORS 를 허용해준다.
  • 추가적으로 기억하면 좋을 것은, FastAPI 에서 Middleware 는 add_middleware 로 추가된 순서의 반대 순서로 호출된다는 점이다.
맨 위로 이동 ↑

댓글 남기기

관련 게시글