Write an Input Adapter¶
from beekeeper import EntityInputAdapter
class CsvWorkerAdapter(EntityInputAdapter[MyWorker]):
def __init__(self, file: Path) -> None:
self._file = file
def get_entities(self) -> Iterable[MyWorker]:
with self._file.open() as f:
for row in csv.DictReader(f):
yield MyWorker(name=row["name"], role=row["role"], unavailabilities=[])
For allocations, subclass AllocationInputAdapter[YourRequest] and implement get_allocations(self) -> Iterable[YourRequest].
Most domains will have one of each. Compose them with CompositeInputAdapter:
adapter = CompositeInputAdapter(
entity_adapter=CsvWorkerAdapter(Path("staff.csv")),
allocation_adapter=ApiAllocationAdapter(api_url="..."),
)
If both kinds of data come from the same source, subclass InputAdapter[TEntity, TAllocReq] directly and implement both methods on one class.
For JSON-from-file, the framework ships JsonEntityInputAdapter and JsonAllocationInputAdapter — strict-only, pydantic-backed, no leniency knob. Use them or write your own.