[throttled-py 기여] 준비과정: 메타클래스와 레지스트리 패턴 이해하기
레지스트리 패턴이란?
- 생성되는 클래스를 자동으로 등록하고 관리하는 디자인 패턴입니다
- 유명한 파이썬 라이브러리에서 많이 씁니다(Django, SQLAlchemy, pydantic, 우리가 기여할 throttled-py, etc.)
flowchart LR
A["class FixedWindow
(BaseRateLimiter):"]
B["RateLimiterMeta
__new__() 가로채기"]
C["RateLimiterRegistry
.register()"]
D["_RATE_LIMITERS
sync:fixed_window → cls"]
E["사용자 코드
Registry.get('fixed_window')"]
A -->|"클래스 정의 시점"| B
B -->|"isinstance 검사 후
등록 대상이면"| C
C -->|"키-클래스 저장"| D
E -->|"문자열로 조회"| D이 글은 위 흐름의 각 단계를 하나씩 뜯어봅니다. 메타클래스가 클래스 정의를 가로채는 원리, 레지스트리에 등록하는 조건, 그리고 사용자가 문자열 하나로 구현체를 꺼내 쓰는 구조까지 순서대로 살펴봅니다.
GoF의 그것과 비교해보기
레지스트리 패턴은 아래 요소들의 조합입니다. 최대한 유사하게 붙여보았습니다.
| GoF 패턴 | 레지스트리에서의 역할 | 예시 |
|---|---|---|
| Factory Method | 이름으로 객체 생성 | Registry.create("token_bucket") |
| Singleton | 전역 레지스트리 관리 | REGISTRY = {} (클래스 변수) |
| Prototype | 클래스 템플릿 저장 | 매번 새 인스턴스 생성 |
| Strategy | 런타임 알고리즘 선택 | using="token_bucket" vs "gcra" |
| Abstract Factory | 관련 객체군 생성 | 알고리즘 + 스토리지 조합 |
| Observer | 등록 시 알림 | 메타클래스가 클래스 정의를 감지 |
요구사항 해제
이번에 살펴볼 throttled-py는 아래 요구사항들을 충족하기 위해 레지스트리 패턴을 선택한 것으로 보입니다.
일단 알고리즘이 최소 3개이상 나오고, 아래 사항을 충족시키는 것이 좋겠습니다:
- 런타임에 고를 수 있으면 좋겠습니다
- 매번 새 인스턴스가 생성되어야 할 겁니다(요청별로나 각각 다르게 쓸 테니)
- 객체 생성이 쉬웠으면 좋겠습니다
그러한 요구사항들을 한번에 녹이기위한 코드 구성이 레지스트리 패턴입니다.
메타클래스 이해하기
레지스트리 패턴을 구현하기 위해 메타클래스라는 프로그래밍 요소를 사용합니다.
메타클래스를 이해하기 전, 클래스와 객체 생성에 대해 다시 한 번 살펴보겠습니다.
일반 클래스에 대해
일반적인 클래스는 인스턴스 객체를 __new__ 로 생성합니다.
일반 클래스의 생성
| 메서드 | 역할 | 받는 것 | 반환 |
|---|---|---|---|
__new__ |
생성자 - 객체 메모리 할당 | cls (클래스) | 인스턴스 객체 |
__init__ |
초기화자 - 속성 설정 | self (인스턴스) | None |
코드를 돌려보면 이건 직관적으로 알 수 있습니다.
In [1]: class Foo:
...: def __new__(cls, *args, **kwargs):
...: print(f"1. __new__: 객체 메모리 할당 (cls={cls})")
...: instance = super().__new__(cls) # 실제 객체 생성
...: print(f"\t할당받은 메모리 주소\n\t{instance}")
...: return instance # 반드시 인스턴스 반환!
...: def __init__(self, name: str):
...: print(f"2. __init__: 객체 초기화")
...: print(f"\t할당받은 메모리 주소의 인스턴스 그대로 사용\n\t{self}")
...: self.name = name
...:
In [2]: foo = Foo("홍길동")
1. __new__: 객체 메모리 할당 (cls=<class '__main__.Foo'>)
할당받은 메모리 주소
<__main__.Foo object at 0x7e6b9eff69f0>
2. __init__: 객체 초기화
할당받은 메모리 주소의 인스턴스 그대로 사용
<__main__.Foo object at 0x7e6b9eff69f0>
도표로 그리면 아래와 같은 형식입니다.
Foo("hello")
│
├─▶ Foo.__new__(cls, "hello")
│ │
│ └─▶ 인스턴스 객체 반환
│
└─▶ Foo.__init__(self, "hello") ← __new__가 반환한 객체가 self로
│
└─▶ self.name = "hello"
메타클래스가 뭔데요?
메타클래스는 '클래스를 만드는 과정을 가로채서 조작하는' 도구입니다. 메타클래스는 여러 클래스에 걸쳐 반복되는 패턴을 자동화하기 한 방법에 활용될 수도 있으며, 파이썬에서는 이를 클래스 "정의" 시점에 클래스 객체를 생성하는 것으로 해결하였습니다.
메타클래스는 다시 강조하자면, 클래스 객체를 생성합니다. 위에서 살펴본 "인스턴스" 생성과는 다릅니다.
일러두기
일단 이걸 이해하기 위해 깔아둬야 할 전제가 있습니다:
파이썬에서는 모든 것이 객체(Object)입니다.
→ 모든 파이썬 클래스도 객체입니다.
모든 객체는 type을 가집니다.
In [1]: class Foo:
...: pass
In [2]: x = Foo()
In [3]: type(x)
Out[3]: __main__.Foo
In [4]: type(Foo)
Out[4]: type
In [5]: print(type(Foo))
Out[5]: <class 'type'>
# type 클래스 객체 자신을 만든 것도 결국 type. 모든 타입 계층의 정점에 있다.
그리고 모든 파이썬 객체의 타입은 type 입니다.
In [1]: for t in int, float, dict, list, tuple:
...: print(type(t))
...:
<class 'type'>
<class 'type'>
<class 'type'>
<class 'type'>
<class 'type'>
type 은 그 자체로 메타클래스이며, 모든 파이썬 클래스는 type 의 인스턴스입니다.
→ 다시말해 모든 클래스는 type 메타클래스의 인스턴스입니다.
In [1]: print(type(type))
<class 'type'>
아래 내용이 이해되지 않는다면, 핵심 링크를 참고하고 직접 디버깅하여 돌려보기를 적극 권장합니다.
메타클래스의 생성
왜 이렇게까지 강조하였냐면, type 이 메타클래스에서 어떻게 활용되는 것인지 이해해야 하기 때문입니다.
예시 코드를 살펴보고, 직접 구동하며 "코드 디버깅 해보기" 과정 대로 이해하길 바랍니다.
In [1]: # 1. 메타클래스 정의 (클래스를 만드는 공장)
...: class MetaC(type):
...: def __new__(mcs, name, bases, attrs):
...: """
...: [시점 1] 파이썬 인터프리터가 'class Bar(...):' 구문을 읽을 때 실행됨.
...: 아직 Bar라는 클래스 객체는 메모리에 없음. 지금 만드는 중.
...: """
...: print(f"\n[1. META __new__] '{name}' 클래스(틀)를 설계도면대로 찍어내는 중...")
...: print(f" -> mcs(메타클래스): {mcs}")
...: print(f" -> attrs(속성들): {list(attrs.keys())}")
...: # type.__new__를 호출하여 실제 '클래스 객체'를 힙 메모리에 생성
...: cls_obj = super().__new__(mcs, name, bases, attrs)
...: print(f" -> [생성 완료] 메모리에 적재된 클래스 객체 주소: {hex(id(cls_obj))}")
...: return cls_obj
...: def __init__(cls, name, bases, attrs):
...: """[시점 2] 클래스 객체가 생성된 직후 초기화"""
...: print(f"[2. META __init__] '{name}' 클래스 객체 초기화 완료")
...: super().__init__(name, bases, attrs)
...: def __call__(cls, *args, **kwargs):
...: """
...: [시점 3] 사용자가 Bar()라고 괄호를 열고 닫을 때 실행됨.
...: 즉, 인스턴스를 만들려고 할 때 '메타클래스'가 먼저 개입함.
...: """
...: print(f"\n[3. META __call__] {cls.__name__}()를 호출함 -> 인스턴스 생성 시작")
...: instance = super().__call__(*args, **kwargs)
...: print(f"[6. META __call__] 인스턴스 생성 및 리턴 완료")
...: return instance
...:
In [2]: # 2. 클래스 정의 — 이 순간 MetaC.__new__ -> MetaC.__init__ 이 실행됨
...: class Bar(metaclass=MetaC):
...: def __new__(cls, *args, **kwargs):
...: print(f"[4. CLASS __new__] {cls.__name__}의 인스턴스(실체) 메모리 할당 중...")
...: instance = super().__new__(cls)
...: print(f"\t[DEBUG] 실제 할당받은 메모리 주소 - {instance}")
...: return instance
...: def __init__(self):
...: print(f"\t[DEBUG] 전달받은 메모리 주소 - {self}")
...: print(f"[5. CLASS __init__] {type(self).__name__} 인스턴스 값 초기화 중")
...:
[1. META __new__] 'Bar' 클래스(틀)를 설계도면대로 찍어내는 중...
-> mcs(메타클래스): <class '__main__.MetaC'>
-> attrs(속성들): ['__module__', '__qualname__', '__new__', '__init__', '__classcell__']
-> [생성 완료] 메모리에 적재된 클래스 객체 주소: 0x2f926240
[2. META __init__] 'Bar' 클래스 객체 초기화 완료
In [3]: hex(id(Bar)) # 위의 [생성 완료] 주소와 동일해야 함
Out[3]: '0x2f926240'
In [4]: # 3. 인스턴스 생성
...: bar = Bar()
[3. META __call__] Bar()를 호출함 -> 인스턴스 생성 시작
[4. CLASS __new__] Bar의 인스턴스(실체) 메모리 할당 중...
[DEBUG] 실제 할당받은 메모리 주소 - <__main__.Bar object at 0x70666fc41af0>
[DEBUG] 전달받은 메모리 주소 - <__main__.Bar object at 0x70666fc41af0>
[5. CLASS __init__] Bar 인스턴스 값 초기화 중
[6. META __call__] 인스턴스 생성 및 리턴 완료
In [5]: bar
Out[5]: <__main__.Bar object at 0x70666fc41af0>
In [6]: type(bar)
Out[6]: <class '__main__.Bar'>
In [7]: type(Bar) # 클래스의 타입 = 메타클래스
Out[7]: <class '__main__.MetaC'>
메타클래스의 흐름 참고하기
- 처음으로 메타클래스 정의를 수행합니다.
- 이 때,
type.__new__(),type.__init__()이 실행되어MetaC클래스 객체가 생성됩니다 - 이후 메타클래스를 사용하기로 한 클래스들의 정보는
name,bases,attrs입니다.name- 클래스 정의문의 이름을 문자열로 처리 (str)bases- 클래스 정의문에 상속하기로 결정된 부모 클래스의 튜플 (tuple)attrs- 클래스 자기자신 만의 속성(attributes)
- 이 때,
- 이어서
Bar클래스의 정의를 확인합니다.- 이후 dunder 메소드들이 뭐가 있는지 체크합니다
- 그런데 클래스 정의에
metaclass=MetaC가 있으면,__new__프로토콜을 실행하러 갑니다.- 이때, Bar 클래스의 정보를 같이 실어갑니다(
name,bases,attrs).
- 이때, Bar 클래스의 정보를 같이 실어갑니다(
MetaC.__new__()가 실행됩니다.cls_obj = super().__new__(mcs, name, bases, attrs)를 구동합니다- 이를 토대로
Bar클래스 객체가 메모리 상에 생성되었습니다.
(MetaC메타클래스 객체가 생성되는 게 아님!!!!!!! 유의)
MetaC.__init__()이 실행됩니다__new__()로 생성한 객체를 받고, 나머지 인자는 모두 동일하게 받습니다- 이를 토대로
Bar클래스 객체를 초기화 합니다.
(MetaC메타클래스 객체를 "초기화"하는 게 아님!!!!!!! 유의)
인스턴스 생성 시?
bar = Bar()로 객체 인스턴스를 생성하면MetaC.__call__()이 호출됩니다Bar.__new__()를 호출해서 인스턴스를 메모리에 생성합니다Bar.__init__()를 호출해서 인스턴스를 "초기화" 합니다.
- 그럼 이렇게 생성된
Bar()객체를 이후엔 사용하는 것입니다.
The Power of Metaclass
그럼 메타클래스가 레지스트리 패턴에서 어떻게 활용되는지 본격적으로 살펴보겠습니다.
아래 코드를 통해 메타클래스가 어떻게 쓰이는지 살펴보겠습니다. 메타클래스의 구현체와, RateLimiterMeta 클래스를 메타클래스로 활용하는 클래스를 보여줍니다.
class RateLimiterMeta(abc.ABCMeta):
"""Metaclass for RateLimiter classes."""
_REGISTRY_CLASS: Type[RateLimiterRegistry] = RateLimiterRegistry
def __new__(cls, name, bases, attrs):
new_cls = super().__new__(cls, name, bases, attrs)
if not [b for b in bases if isinstance(b, cls)]:
return new_cls
cls._REGISTRY_CLASS.register(new_cls)
return new_cls
# 해당 클래스가 로드되는 시점에
# RateLimiterMeta 메타클래스가 BaseRateLimiter 클래스 객체를 생성한다
class BaseRateLimiter(BaseRateLimiterMixin, metaclass=RateLimiterMeta):
"""Base class for RateLimiter."""
...
# 마찬가지로, 이 클래스가 로드되는 시점에
# RateLimiterMeta 클래스가 FixedWindowRateLimiter 클래스 객체를 생성한다
class FixedWindowRateLimiter(FixedWindowRateLimiterCoreMixin, BaseRateLimiter):
"""Concrete implementation of BaseRateLimiter using fixed window as algorithm."""
음.... 이렇게 봐선 아무래도 어렵군요 😅
디버깅해보기
잘 모를 때는 역시 디버깅입니다. 디버깅을 진행함으로써 왜 이렇게 되나 살펴보겠습니다.
-
코드베이스의 테스트코드를 구동해서 디버그 해보세요.
import순서대로 어떻게 메타클래스가__new__()로 생성되고,BaseRateLimiter가 로드될 때 클래스를 로드하는지 볼 수 있습니다. -
아래 글에서 🔴 는 브레이크포인트를 의미합니다.
1. 객체 호출
Throttled() 객체를 호출합니다.
2. __init__.py 로드
파이썬에선 라이브러리 import 를 수행하면 라이브러리의 __init__.py 부터 호출하며, 다른 모듈들을 로드합니다.
# /throttled/__init__.py
"""High-performance Python rate limiting library."""
from . import asyncio, constants, exceptions, rate_limiter, types, utils
3. 동일 경로의 asyncio 모듈 로드
이를 로드하면서 라이브러리 디렉터리의 asyncio를 먼저 import 합니다.
# /throttled/asyncio/__init__.py
"""Asyncio support for throttled."""
🔴from .. import constants, exceptions, rate_limiter, types, utils
4. rate_limiter의 __init__.py 로드
이후 임포트 순서대로(좌에서 우로) throttled.rate_limiter.__init__ 을 로드합니다.
from .base import (
BaseRateLimiter,
BaseRateLimiterMixin,
Quota,
Rate,
RateLimiterMeta,
RateLimiterRegistry,
RateLimitResult,
RateLimitState,
per_day,
per_duration,
per_hour,
per_min,
per_sec,
per_week,
)
# Trigger to register RateLimiter
from .fixed_window import FixedWindowRateLimiter
5. 동기 버전의 RateLimiter 를 호출
하면서 동기 버전의 RateLimiter를 트리거합니다. 즉 레지스트리 클래스 객체에 현재 클래스 객체를 "등록"한다는 것을 의미합니다.
이후
# /throttled/asyncio/__init__.py
"""Asyncio support for throttled."""
# 여기를 마쳤으니
from .. import constants, exceptions, rate_limiter, types, utils
from ..constants import RateLimiterType
# 여기 로드를 시작한다.
🔴 from .rate_limiter import (
BaseRateLimiter,
Quota,
Rate,
RateLimiterMeta,
RateLimiterRegistry,
RateLimitResult,
RateLimitState,
per_day,
per_duration,
per_hour,
per_min,
per_sec,
per_week,
)
6. 비동기 버전의 로직을 쫓아가기
로드하는 파일은 throttled.asyncio.rate_limiter.__init__.py 이니까, 거기서부터 다시 로드하면, 여기서부터 시작합니다.
# 위의 로직은 이미 캐시되어있으니 상관없다
from ...rate_limiter.base import (
Quota,
Rate,
RateLimitResult,
RateLimitState,
per_day,
per_duration,
per_hour,
per_min,
per_sec,
per_week,
)
# 여기서부터 로드!
🔴 from .base import BaseRateLimiter, RateLimiterMeta, RateLimiterRegistry
# Trigger to register Async RateLimiter
from .fixed_window import FixedWindowRateLimiter
from .gcra import GCRARateLimiterCoreMixin
from .leaking_bucket import LeakingBucketRateLimiter
from .sliding_window import SlidingWindowRateLimiter
from .token_bucket import TokenBucketRateLimiter
7. base.py 파일을 메모리에 로드하기
그러면 base 부터 다시 읽으며 클래스 변수도 생성합니다.
# /throttled/asyncio/rate_limiter/base.py
import abc
from typing import Type
from ... import rate_limiter
class RateLimiterRegistry(rate_limiter.RateLimiterRegistry):
"""Registry for Async RateLimiter classes."""
_NAMESPACE: str = "async"
class RateLimiterMeta(rate_limiter.RateLimiterMeta):
"""Metaclass for Async RateLimiter classes."""
_REGISTRY_CLASS: Type[RateLimiterRegistry] = RateLimiterRegistry
class BaseRateLimiter(rate_limiter.BaseRateLimiterMixin, metaclass=RateLimiterMeta):
"""Base class for Async RateLimiter."""
@abc.abstractmethod
async def _limit(self, key: str, cost: int) -> rate_limiter.RateLimitResult:
raise NotImplementedError
@abc.abstractmethod
async def _peek(self, key: str) -> rate_limiter.RateLimitState:
raise NotImplementedError
async def limit(self, key: str, cost: int = 1) -> rate_limiter.RateLimitResult:
return await self._limit(key, cost)
async def peek(self, key: str) -> rate_limiter.RateLimitState:
return await self._peek(key)
이러면 RateLimiterRegistry는 rate_limiter.RateLimiterRegistry 를 상속하게 되는데, 이러면 기존에 선언한 값을 런타임에 동일한 이름의 클래스로 모두 상속할 수 있습니다.
이후 여기서도 기존에 동기로직에서 학습했던 것과 마찬가지로 metaclass 를 통해 동일한 로직을 거칩니다.
# 위의 로직은 이미 캐시되어있으니 상관없다
from .base import BaseRateLimiter, RateLimiterMeta, RateLimiterRegistry
# Trigger to register Async RateLimiter
🔴 from .fixed_window import FixedWindowRateLimiter
from .gcra import GCRARateLimiterCoreMixin
from .leaking_bucket import LeakingBucketRateLimiter
from .sliding_window import SlidingWindowRateLimiter
from .token_bucket import TokenBucketRateLimiter
이 커맨드를 수행하면 모듈 로드 완료 시점마다 import 'module.name' 을 찍어줍니다. 이를 우리가 살펴보려는 라이브러리로 볼 수 있습니다.
uv run python -v -c "import throttled" 2>&1 | grep "^import 'throttled"
$ uv run python -v -c "import throttled" 2>&1 | grep "^import 'throttled"
import 'throttled.types' # <_frozen_importlib_external.SourceFileLoader object at 0x7f24391f3c10>
import 'throttled.constants' # <_frozen_importlib_external.SourceFileLoader object at 0x7f24391ae750>
import 'throttled.exceptions' # <_frozen_importlib_external.SourceFileLoader object at 0x7f24391aeed0>
import 'throttled.rate_limiter.base' # <_frozen_importlib_external.SourceFileLoader object at 0x7f2438981cd0>
import 'throttled.store.base' # <_frozen_importlib_external.SourceFileLoader object at 0x7f24388e9810>
import 'throttled.utils' # <_frozen_importlib_external.SourceFileLoader object at 0x7f24388ebad0>
import 'throttled.store.memory' # <_frozen_importlib_external.SourceFileLoader object at 0x7f24388ead90>
import 'throttled.store.redis_pool' # <_frozen_importlib_external.SourceFileLoader object at 0x7f24386f1e50>
import 'throttled.store.redis' # <_frozen_importlib_external.SourceFileLoader object at 0x7f24386df750>
import 'throttled.store' # <_frozen_importlib_external.SourceFileLoader object at 0x7f24388e90d0>
import 'throttled.rate_limiter.fixed_window' # <_frozen_importlib_external.SourceFileLoader object at 0x7f24388e8ad0>
import 'throttled.rate_limiter.gcra' # <_frozen_importlib_external.SourceFileLoader object at 0x7f243854d610>
import 'throttled.rate_limiter.leaking_bucket' # <_frozen_importlib_external.SourceFileLoader object at 0x7f243854d690>
import 'throttled.rate_limiter.sliding_window' # <_frozen_importlib_external.SourceFileLoader object at 0x7f243854fc10>
import 'throttled.rate_limiter.token_bucket' # <_frozen_importlib_external.SourceFileLoader object at 0x7f2438564d90>
import 'throttled.rate_limiter' # <_frozen_importlib_external.SourceFileLoader object at 0x7f243895e8d0>
import 'throttled.asyncio.rate_limiter.base' # <_frozen_importlib_external.SourceFileLoader object at 0x7f24385661d0>
import 'throttled.asyncio.store.base' # <_frozen_importlib_external.SourceFileLoader object at 0x7f2438567210>
import 'throttled.asyncio.store.memory' # <_frozen_importlib_external.SourceFileLoader object at 0x7f2438567d50>
import 'throttled.asyncio.store.redis' # <_frozen_importlib_external.SourceFileLoader object at 0x7f243857c850>
import 'throttled.asyncio.store' # <_frozen_importlib_external.SourceFileLoader object at 0x7f2438566c50>
import 'throttled.asyncio.rate_limiter.fixed_window' # <_frozen_importlib_external.SourceFileLoader object at 0x7f2438566810>
import 'throttled.asyncio.rate_limiter.gcra' # <_frozen_importlib_external.SourceFileLoader object at 0x7f243857d410>
import 'throttled.asyncio.rate_limiter.leaking_bucket' # <_frozen_importlib_external.SourceFileLoader object at 0x7f243857da10>
import 'throttled.asyncio.rate_limiter.sliding_window' # <_frozen_importlib_external.SourceFileLoader object at 0x7f243857df50>
import 'throttled.asyncio.rate_limiter.token_bucket' # <_frozen_importlib_external.SourceFileLoader object at 0x7f243857e490>
import 'throttled.asyncio.rate_limiter' # <_frozen_importlib_external.SourceFileLoader object at 0x7f2438565050>
import 'throttled.hooks' # <_frozen_importlib_external.SourceFileLoader object at 0x7f243857fed0>
import 'throttled.throttled' # <_frozen_importlib_external.SourceFileLoader object at 0x7f243857f050>
import 'throttled.asyncio.throttled' # <_frozen_importlib_external.SourceFileLoader object at 0x7f2438981990>
import 'throttled.asyncio' # <_frozen_importlib_external.SourceFileLoader object at 0x7f24391ae450>
import 'throttled' # <_frozen_importlib_external.SourceFileLoader object at 0x7f2439111290>
8. rate_limiter/base.py 로드 - 클래스 객체 생성
throttled/rate_limiter/__init__.py 의 import 구문을 통해 BaseRateLimiter를 import 하며 클래스 객체들을 각각 생성합니다.
import abc
import logging
from dataclasses import dataclass
from datetime import timedelta
from typing import Dict, List, Optional, Set, Tuple, Type
from ..exceptions import SetUpError
from ..types import AtomicActionP, AtomicActionTypeT, RateLimiterTypeT, StoreP
logger: logging.Logger = logging.getLogger(__name__)
@dataclass
class Rate:
...
# 이후 아래에 있는 모든 메소드를 파악하고 클래스 객체를 생성한다.
9. BaseRateLimiter 정의 시 - 메타클래스 __new__() 호출
class BaseRateLimiter(BaseRateLimiterMixin, metaclass=RateLimiterMeta): 를 만났을 땐 얘기가 다릅니다. 메타클래스가 지정되어있으니, RateLimiterMeta 의 __new__() 를 수행합니다.
→ [[#BaseRateLimiter 정의 시]] 문단을 참고하시길 바랍니다.
BaseRateLimiter클래스 객체가 생성되는 과정을 낚아채서 추가적인 작업을 수행하게 한 것입니다.- 이 때 레지스트리 클래스에 값을 넣을지 말지에 대한 로직을 탑니다.
10. FixedWindowRateLimiter import 트리거
8번 구문이 로드 되었으니, 아래 내용을 다시 살펴보겠습니다.
# Trigger to register RateLimiter
from .fixed_window import FixedWindowRateLimiter
11. fixed_window.py 로드 - 클래스 객체 생성
마찬가지로 throttled/rate_limiter/__init__.py 의 import 구문을 통해 FixedWindowRateLimiter 를 import 하며 클래스 객체들을 각각 생성합니다.
from typing import TYPE_CHECKING, List, Optional, Sequence, Tuple, Type
from ..constants import ATOMIC_ACTION_TYPE_LIMIT, RateLimiterType, StoreType
from ..store import BaseAtomicAction
from ..types import AtomicActionP, AtomicActionTypeT, KeyT, RateLimiterTypeT, StoreValueT
from ..utils import now_sec
from . import BaseRateLimiter, BaseRateLimiterMixin, RateLimitResult, RateLimitState
if TYPE_CHECKING:
from ..store import MemoryStoreBackend, RedisStoreBackend
class RedisLimitAtomicActionCoreMixin:
"""Core mixin for RedisLimitAtomicAction."""
...
# 이후 아래에 있는 클래스 객체를 생성한다.
12. FixedWindowRateLimiter 정의 시 - 메타클래스 __new__() 호출
여기서도, class FixedWindowRateLimiter(FixedWindowRateLimiterCoreMixin, BaseRateLimiter): 를 만났을 땐 얘기가 다릅니다. BaseRateLimiter는 메타클래스를 상속한 것이니, RateLimiterMeta 의 __new__() 를 수행합니다.
→ [[#FixedWindowRateLimiter 정의 시]] 문단을 참고하시길 바랍니다.
FixedWindowRateLimiter클래스 객체가 생성되는 과정을 낚아채서 추가적인 작업을 수행하게 한 것입니다.- 이 때 레지스트리 클래스에 값을 넣을지 말지에 대한 로직을 탑니다.
위에서 살펴보았던 것 처럼 isinstance(BaseRateLimiter, RateLimiterMeta) 가 True인 이유는 클래스도 객체이고, BaseRateLimiter라는 클래스 객체를 만든 메타클래스가 RateLimiterMeta이기 때문입니다.
필요한 요소만 레지스트리 클래스에 등록(register)하기
그렇다면, 레지스트리 패턴을 이용하여 메타클래스 내에서 클래스를 등록하는 부분에 대해 살펴보겠습니다.
class RateLimiterMeta(abc.ABCMeta):
"""Metaclass for RateLimiter classes."""
_REGISTRY_CLASS: Type[RateLimiterRegistry] = RateLimiterRegistry
def __new__(cls, name, bases, attrs):
new_cls = super().__new__(cls, name, bases, attrs)
# bases 중에 RateLimiterMeta로 만든 클래스가 하나라도 있는지 확인
# → 즉, "이 메타클래스 시스템에 속한 부모를 가졌나?"
if not [b for b in bases if isinstance(b, cls)]:
# 없으면 → 기반 클래스(BaseRateLimiter)이므로 등록 안 함
return new_cls
# 있으면 → 실제 구현체(FixedWindowRateLimiter 등)이므로 등록함
cls._REGISTRY_CLASS.register(new_cls)
return new_cls
isinstance 는 어떤 객체가 어떤 클래스의 "인스턴스" 인지 물어보는 메소드입니다.
object가classinfo의 인스턴스이거나,object가classinfo의 자식 클래스의 인스턴스이면
True를, 그렇지 않으면 False 를 리턴합니다.
방금전까지 클래스 또한 객체이며, 이 클래스를 만드는 클래스가 메타클래스임을 살펴보고 왔습니다. 즉, 어느 클래스로 인해 클래스 인스턴스가 생성되었는지 판별할 수 있다는 소리입니다. 다시말해 클래스 시그니처에서, metaclass=어쩌고 로 메타클래스가 동일하거나, 그 하위 메타클래스인지 물어보는 로직이라 할 수 있습니다.
그렇다면 위에서 살펴본 import 순서대로, 아래 클래스들의 메타클래스 __new__() 가 작동하는 방식을 살펴보겠습니다.
BaseRateLimiter 정의 시
BaseRateLimiter정의 시class BaseRateLimiter(BaseRateLimiterMixin, metaclass=RateLimiterMeta):bases = (BaseRateLimiterMixin,)가 들어오고- 반복문을 돌자.
isinstance(BaseRateLimiterMixin, RateLimiterMeta)→BaseRateLimiterMixin의 메타클래스는type이므로 →False
- 그 결과 리스트가
[]→not []이므로True를 리턴하니, 레지스터를 하지 않습니다.결과:
BaseRateLimiter는 레지스터 안에 소속되지 않습니다
FixedWindowRateLimiter 정의 시
FixedWindowRateLimiter정의 시class FixedWindowRateLimiter(FixedWindowRateLimiterCoreMixin, BaseRateLimiter):bases = (FixedWindowRateLimiterCoreMixin, BaseRateLimiter)가 파라미터로 들어오고- 반복문을 돌자.
isinstance(FixedWindowRateLimiterCoreMixin, RateLimiterMeta)→ 믹스인의 메타클래스는type이므로 →Falseisinstance(BaseRateLimiter, RateLimiterMeta)→BaseRateLimiter는metaclass=RateLimiterMeta로 만들어진 클래스 객체다.BaseRateLimiter는RateLimiterMeta의 인스턴스이므로 →True
- 그 결과 리스트가
[BaseRateLimiter]→not [BaseRateLimiter]이므로False를 리턴하니 레지스터를 합니다.결과:
FixedWindowRateLimiter는 레지스터 안에 소속됩니다
반면 Mixin 들은 type 이 메타클래스니까 해당되지 않습니다. BaseRateLimiter 는 메타클래스를 metaclass=RateLimiterMeta 으로 지정해주었으므로, 상속하는 실제 구현체 FixedWindowRateLimiter 등은 레지스트리에 등록됩니다.
(메타클래스는 상속됨 → 자식도 RateLimiterMeta로 생성됨)
추가로:
RateLimiterMeta(abc.ABCMeta) 에서 abc.ABCMeta 를 상속하는 이유는 BaseRateLimiter에서 @abc.abstractmethod를 사용하기 위함입니다. 만약 type을 직접 상속했다면 추상 메소드 강제가 작동하지 않습니다.
레지스트리 클래스 이해하기
이해를 위해 우선 동기 코드만 살펴봅니다. 비동기 코드는 필요한 부분만 언급하고 나머지는 다음에 설명하도록 합니다.
우리는 위의 과정을 거쳐 여러 클래스에 걸쳐 반복되는 패턴을 자동화 했습니다. 이러면 위에서 말한 대로 RateLimiter 구현체들은 필요에 따라 _REGISTRY_CLASS 에 등록(register)하고 필요할 때 쓸 수 있도록 구성했습니다.
그렇다면 RateLimiterRegistry의 코드를 보고 정확히 어떻게 쓰이는지도 살펴보겠습니다:
# /throttled/rate_limiter/base.py
class RateLimiterRegistry:
"""Registry for RateLimiter classes."""
# The namespace for the RateLimiter classes.
_NAMESPACE: str = "sync"
# A dictionary to hold the registered RateLimiter classes.
_RATE_LIMITERS: Dict[RateLimiterTypeT, Type["BaseRateLimiter"]] = {}
@classmethod
def get_register_key(cls, _type: str) -> str:
"""Get the register key for the RateLimiter classes."""
return f"{cls._NAMESPACE}:{_type}"
@classmethod
def register(cls, new_cls):
try:
cls._RATE_LIMITERS[cls.get_register_key(new_cls.Meta.type)] = new_cls
except AttributeError as e:
raise SetUpError("failed to register RateLimiter: {}".format(e))
@classmethod
def get(cls, _type: RateLimiterTypeT) -> Type["BaseRateLimiter"]:
try:
return cls._RATE_LIMITERS[cls.get_register_key(_type)]
except KeyError:
raise SetUpError("{} not found".format(_type))
_NAMESPACE: 네임스페이스str. 동기 버전은sync, 비동기 버전은async로 클래스 변수를 오버라이드합니다_RATE_LIMITERS: 실제로 다양한RateLimiter구현체(클래스)를 담기 위한dict입니다. 키는 타입명, 밸류는 실제 클래스get_register_key(_type): 레지스트리에 저장하려는 키를 만들어줍니다.f"{네임스페이스}:{타입}"register(new_cls): 구현체를 저장합니다.get(_type): 클래스 객체를 꺼내줍니다(인스턴스를 주는 게 아닙니다)
이 클래스도 앞서 살펴보았듯 클래스 객체로서 존재합니다.
또한 _NAMESPACE, _RATE_LIMITERS 라는 이름의 변수를 바로 쓸 수 있는 이유는, 저런 식으로 변수를 초기화 하는 건 클래스 변수정의를 한 것이기 때문입니다. (https://docs.python.org/3/tutorial/classes.html#class-and-instance-variables)
상술하였듯 isinstance 를 통해 동일한 메타클래스로부터 생성되었는지 점검 후, 실제 구현체만 cls._REGISTRY_CLASS.register() 로 클래스메소드로 등록해줍니다. 이러면 유연하게 등록이 가능합니다.
비동기 버전의 RateLimiterRegistry 는 아래와같이 sync 버전을 상속합니다. _RATE_LIMITERS 를 재정의하지 않았으므로 동일 dict 객체를 공유하고, _NAMESPACE 는 값을 다르게 쓰기 때문에 가능합니다.
# /throttled/asyncio/rate_limiter/base.py
class RateLimiterRegistry(rate_limiter.RateLimiterRegistry):
"""Registry for Async RateLimiter classes."""
_NAMESPACE: str = "async"
추가로:
클래스메소드(@classmethod) 는 그래서 객체 생성없이 바로 쓸 수 있는 것입니다. 일반 메소드였다면 클래스로부터 생성한 인스턴스 객체 별로 초기화 후 메소드를 사용해야했던 것도 이런 이유 때문입니다.
Meta 의 정체
register 메소드의 주석 부분을 조금 더 자세히 보겠습니다. 뜬금없이 new_cls 의 Meta 를 참조하고 있습니다.
@classmethod
def register(cls, new_cls):
try:
# 여기의 Meta란?
# 여기 브레이크포인트를 걸었다 칩시다
🔴 cls._RATE_LIMITERS[cls.get_register_key(new_cls.Meta.type)] = new_cls
except AttributeError as e:
raise SetUpError("failed to register RateLimiter: {}".format(e))
그렇다면 저 Meta 는 뭘 말하는 걸까요? 이건 클래스 상속 이후 객체를 찾는 개념입니다.
다시말해, 상속구조 상의 CoreMixin 안의 클래스, Meta 를 의미합니다.
class FixedWindowRateLimiterCoreMixin(BaseRateLimiterMixin):
"""Core mixin for FixedWindowRateLimiter."""
_DEFAULT_ATOMIC_ACTION_CLASSES: List[Type[AtomicActionP]] = []
class Meta:
type: RateLimiterTypeT = RateLimiterType.FIXED_WINDOW.value
@classmethod
def _default_atomic_action_classes(cls) -> List[Type[AtomicActionP]]:
return cls._DEFAULT_ATOMIC_ACTION_CLASSES
@classmethod
def _supported_atomic_action_types(cls) -> List[AtomicActionTypeT]:
return [ATOMIC_ACTION_TYPE_LIMIT]
def _prepare(self, key: str) -> Tuple[str, int, int, int]:
now: int = now_sec()
period: int = self.quota.get_period_sec()
period_key: str = f"{key}:period:{now // period}"
return self._prepare_key(period_key), period, self.quota.get_limit(), now
객체 꺼내오기
레지스트리 변수로 저장된 실제 클래스를 확인하는 것은 레지스트리 클래스의 클래스메소드인 .get() 으로 키값만 기재해주면 객체를 바로 꺼내올 수 있습니다.
단순히 MRO 대로 찾아오는 것입니다.
In [11]: new_cls.__mro__
Out[11]:
(throttled.rate_limiter.fixed_window.FixedWindowRateLimiter,
throttled.rate_limiter.fixed_window.FixedWindowRateLimiterCoreMixin,
throttled.rate_limiter.base.BaseRateLimiter,
throttled.rate_limiter.base.BaseRateLimiterMixin,
object)
In [12]: new_cls.__mro__[1]
Out[12]: throttled.rate_limiter.fixed_window.FixedWindowRateLimiterCoreMixin
In [13]: new_cls.__mro__[1].Meta
Out[13]: throttled.rate_limiter.fixed_window.FixedWindowRateLimiterCoreMixin.Meta
In [14]: new_cls.__mro__[1].Meta.type
Out[14]: 'fixed_window'
In [16]: cls.get_register_key(new_cls.Meta.type)
Out[16]: 'sync:fixed_window'
다시말해, 아래와 같은 의미입니다.
cls._RATE_LIMITERS['sync:fixed_window'] = new_cls
요약
메타클래스를 활용한 레지스트리 패턴을 통해 아래 이점을 얻을 수 있습니다:
- 새로운 요구사항 추가도 유연하게 할 수 있고(새 알고리즘 생성 시 이에 대한 내용을 메타클래스로 풀어버리니까)
- 쓰는 것도 이미 약속된 문자열로 쉽게 쓸 수 있습니다. 이게 되니까 모든 알고리즘을 일일이 사용자가 임포트할 필요가 없습니다
graph TB
subgraph meta["메타클래스 레이어"]
RLMeta["RateLimiterMeta
(abc.ABCMeta 상속)
자동 등록 메타클래스"]
end
subgraph base["기반 클래스 레이어"]
Mixin["BaseRateLimiterMixin
quota, store, atomic 공통 로직"]
BRL["BaseRateLimiter
metaclass=RateLimiterMeta
limit(), peek() 인터페이스"]
end
subgraph impl["구현체 — 클래스 정의 시 자동 등록"]
FW["FixedWindow
Meta.type = fixed_window"]
TB2["TokenBucket
Meta.type = token_bucket"]
SW["SlidingWindow
Meta.type = sliding_window"]
GC["GCRA
Meta.type = gcra"]
end
subgraph registry["레지스트리"]
RLR["RateLimiterRegistry
_RATE_LIMITERS: dict
_NAMESPACE: sync | async"]
end
RLMeta -->|"metaclass"| BRL
Mixin -->|"mixin 상속"| BRL
BRL -->|"extends"| FW & TB2 & SW & GC
FW & TB2 & SW & GC -->|"register()"| RLRthrottled-py 에 기여하기 위해 이런 결론을 낼 수 있었습니다.
이 레이어에서 유의미한 메트릭을 수집 가능한 위치는 아래와 같습니다:
limit()메서드 (268라인): 요청 카운터, 레이턴시RateLimitResult생성 시: 허용/거부 비율_prepare_key(): 키별 메트릭 그룹핑