Here is "loadable.py":

from \_\_future\_\_ import annotations

from typing import TYPE\_CHECKING, Generic, Iterable, Mapping, Sequence, TypeVar

import attr

from . import cache

if TYPE\_CHECKING:
 from . import Loader

T = TypeVar('T')
@attr.s(auto\_attribs=True, kw\_only=True, frozen=True, slots=True)
class LoadableConfig:
 """
 Configuration parameters for a Loadable, used by its Loader.
 """

 chunk\_size: int = 100
 """
 The maximum number of ids passed to the \_load\_func.
 """
 max\_request\_workers: int = 4
 """
 The maximum number of concurrent requests to the \_load\_func.
 """
class Loadable(Generic[T]):
 """
 Supertype for all Loader-enabled classes.

 Example:
 class Address(Loadable):
 ...

 Address.load(address\_ids) # preload addresses into cache to prevent N+1s
 address = Address.get\_or\_raise(address\_ids[0]) # single address retrieval, calls \_load\_func if not cached

 A type (like Address) becomes "loadable" by subclassing Loadable, and
 implementing the \_load\_func.

 Each Loadable is associated with a single Loader within the registered
 RequestCache, which is a thread-local dict in session.info by default.
 https://docs.sqlalchemy.org/en/14/orm/session\_api.html#sqlalchemy.orm.Session.info
 """

 \_loadable\_config = LoadableConfig()
 """
 The default config for all Loadables. Override in subclasses as needed.
 """

 @classmethod
 def load(cls: type[T] | type[Loadable], ids: Iterable[str], /) -> None:
 """
 Request preloading for the given ids, calling \_load\_func if necessary.
 """
 cls.\_loader().load(ids)

 @classmethod
 def get(cls: type[T] | type[Loadable], id: str, /) -> T | None:
 """
 Fetch an instance from the Loader cache, fallback to using \_load\_func.

 Returns None if no instance was loaded for the given id.
 """
 return cls.\_loader().get(id)

 @classmethod
 def maybe\_get\_or\_raise(cls: type[T] | type[Loadable], id: str | None, /) -> T | None:
 """
 Calls get\_or\_raise only when the id is non-None, saving an if check.
 """
 if id is None:
 return None
 return cls.get\_or\_raise(id)

 @classmethod
 def get\_or\_raise(cls: type[T] | type[Loadable], id: str, /) -> T:
 """
 Calls get, and raises ValueError if no instance was loaded.
 """
 item = cls.\_loader().get(id)
 if item is None:
 raise ValueError(f'failed to get id {id} for loadable {cls.\_\_name\_\_}')
 return item

 @classmethod
 def bulk\_get(cls: type[T] | type[Loadable], ids: Iterable[str], /) -> Sequence[T]:
 """
 Fetch instances from the Loader cache, fallback to using \_load\_func.

 This method only returns loaded instances in an arbitrary order. Perhaps
 a bulk\_get\_or\_raise addition is warranted.
 """
 return cls.\_loader().bulk\_get(ids)

 @classmethod
 def purge(cls: type[T] | type[Loadable], ids: Iterable[str], /) -> None:
 """
 Remove an instance from the Loader cache.

 By default, purges ids are reloaded on the next \_load\_func invocation to
 help prevent N+1s. Note that performing a system-wide purge, by using
 reit.lib.loader.purge() will NOT trigger this reloading behavior.
 """
 cls.\_loader().purge(ids)

 @classmethod
 def has(cls: type[T] | type[Loadable], id, /) -> bool:
 """
 Check if an instance is already in the Loader cache.

 This method does NOT return True for unprocessed ids from "load" calls.
 """
 return cls.\_loader().has(id)

 @classmethod
 def put(cls: type[T] | type[Loadable], id: str, instance: T, /) -> None:
 """
 Saved an instance into the Loader cache.
 """
 cls.\_loader().put(id, instance)

 @classmethod
 def \_load\_func(cls, ids: Sequence[str], /) -> Mapping[str, T | None]:
 """
 Fetch instances for the given ids.

 A key-value must be returned for each id. A value of None indicates that
 no instance was loaded.
 """
 raise NotImplementedError

 @classmethod
 def \_loader(cls: type[T] | type[Loadable]) -> Loader[T]:
 """
 Obtain (or construct) the Loader for this Loadable.
 """
 return cache.loader\_from\_loadable(cls, cls.\_loadable\_config)
다음은 "loadable.py"입니다:from \_\_future\_\_ 어노테이션 임포트유형에서 TYPE\_CHECKING, Generic, 이터러블, 매핑, 시퀀스, TypeVar를 임포트합니다.import attr에서 . 캐시 임포트if TYPE\_CHECKING: 에서 . 임포트 로더T = TypeVar('T')@attr.s(auto\_attribs=True, kw\_only=True, frozen=True, slots=True)LoadableConfig 클래스: """ 로더가 사용하는 로더블의 환경설정 파라미터입니다. """ 청크\_size: int = 100 """ load\_func에 전달되는 최대 ID 수입니다. """ 최대\_요청\_작업자: int = 4 """ load\_func에 대한 최대 동시 요청 수입니다. """Loadable(Generic[T]) 클래스: """ 모든 로더 사용 가능 클래스의 수퍼형. 예제: 클래스 Address(Loadable): ... Address.load(address\_ids) # N+1초를 방지하기 위해 캐시에 주소를 미리 로드합니다. 주소 = Address.get\_or\_raise(address\_ids[0]) # 단일 주소 검색, 캐시되지 않은 경우 \_load\_func 호출 (Address와 같은) 형은 Loadable을 서브클래싱하고 로드\_func를 구현하면 "로드 가능"이 됩니다. 각 Loadable은 등록된 단일 로더와 연관됩니다. 세션.info의 스레드-로컬 딕셔너리인 RequestCache에 기본적으로 등록됩니다. https://docs.sqlalchemy.org/en/14/orm/session\_api.html#sqlalchemy.orm.Session.info """ \_loadable\_config = LoadableConfig() """ 모든 로더블에 대한 기본 구성입니다. 필요에 따라 서브클래스에서 재정의합니다. """ @classmethod def load(cls: type[T] | type[Loadable], ids: Iterable[str], /) -> None: """ 주어진 id에 대한 사전 로딩을 요청하고, 필요한 경우 \_load\_func를 호출합니다. """ cls.\_loader().load(ids) 클래스 메서드 def get(cls: type[T] | type[Loadable], id: str, /) -> T | None: """ 로더 캐시에서 인스턴스를 가져오고, \_load\_func를 사용하는 것으로 폴백합니다. 주어진 id에 대한 인스턴스가 로드되지 않으면 None을 반환합니다. """ 반환 cls.\_loader().get(id) 클래스 메서드 def maybe\_get\_or\_raise(cls: type[T] | type[Loadable], id: str | None, /) -> T | None: """ id가 None이 아닌 경우에만 get\_or\_raise를 호출하여 if 검사를 절약합니다. """ 를 반환합니다: None을 반환합니다. return cls.get\_or\_raise(id) 클래스 메서드 def get\_or\_raise(cls: type[T] | type[Loadable], id: str, /) -> T: """ get을 호출하고 인스턴스가 로드되지 않은 경우 ValueError를 발생시킵니다. """ item = cls.\_loader().get(id) 항목이 None인 경우: raise ValueError(f'로드 가능한 {cls.\_\_name\_\_}에 대해 {id}를 가져올 수 없습니다') 반환 항목 클래스 메서드 def bulk\_get(cls: type[T] | type[Loadable], ids: Iterable[str], /) -> Sequence[T]: """ 로더 캐시에서 인스턴스를 가져오고, \_load\_func를 사용하는 것으로 폴백합니다. 이 메서드는 로드된 인스턴스를 임의의 순서로만 반환합니다. 아마도 벌크\_get\_또는\_raise 추가가 필요할 수 있습니다. """ 반환 cls.\_loader().bulk\_get(ids) @classmethod def purge(cls: type[T] | type[Loadable], ids: Iterable[str], /) -> None: """ 로더 캐시에서 인스턴스를 제거합니다. 기본적으로, 제거된 id는 다음 \_load\_func 호출 시 다시 로드되어 N+1을 방지합니다. 시스템 전체에 퍼지를 수행하려면 reit.lib.loader.purge()를 사용하면 이 재로드 동작이 트리거되지 않습니다. """ cls.\_loader().purge(ids) 클래스 메서드 def has(cls: type[T] | type[Loadable], id, /) -> bool: """ 인스턴스가 이미 로더 캐시에 있는지 확인합니다. 이 메서드는 "load" 호출에서 처리되지 않은 ID에 대해서는 True를 반환하지 않습니다. """ 반환 cls.\_loader().has(id) 클래스 메서드 def put(cls: type[T] | type[Loadable], id: str, instance: T, /) -> None: """ 인스턴스를 로더 캐시에 저장했습니다. """ cls.\_loader().put(id, instance) 클래스 메서드 def \_load\_func(cls, ids: Sequence[str], /) -> Mapping[str, T | None]: """ 주어진 id에 대한 인스턴스를 가져옵니다. 각 id에 대해 키-값을 반환해야 합니다. None 값은 인스턴스가 로드되지 않았음을 나타냅니다. """ NotImplementedError 발생 클래스 메서드 def \_loader(cls: type[T] | type[Loadable]) -> Loader[T]: """ 이 로더블에 대한 로더를 가져옵니다(또는 구축합니다). """ 반환 cache.loader\_from\_loadable(cls, cls.\_loadable\_config)