Class BaseApiManager

app/api/v2/managers/base_api_manager.py:16–156

class BaseApiManager(BaseWorld):
    def __init__(self, data_svc, file_svc, logger=None):
        self._data_svc = data_svc
        self._file_svc = file_svc
        self._log = logger or self._create_default_logger()

    @property
    def log(self):
        return self._log

    def find_objects(self, ram_key: str, search: dict = None):
        """Find objects matching the given criteria"""
        for obj in self._data_svc.ram[ram_key]:
            if not search or obj.match(search):
                yield obj

    def find_object(self, ram_key: str, search: dict = None):
        for obj in self.find_objects(ram_key, search):
            return obj

    def find_and_dump_objects(self, ram_key: str, search: dict = None, sort: str = None, include: List[str] = None,
                              exclude: List[str] = None):
        matched_objs = []
        for obj in self.find_objects(ram_key, search):
            dumped_obj = self.dump_object_with_filters(obj, include, exclude)
            matched_objs.append(dumped_obj)
        sorted_objs = sorted(matched_objs, key=lambda p: p.get(sort, 0))
        if sorted_objs and sort in sorted_objs[0]:
            return sorted(sorted_objs,
                          key=lambda x: 0 if x[sort] == self._data_svc.get_config(f"objects.{ram_key}.default") else 1)
        return sorted_objs

    @staticmethod
    def dump_object_with_filters(obj: Any, include: List[str] = None, exclude: List[str] = None) -> dict:
        dumped = obj.display
        if include:
            exclude_attributes = list(set(dumped.keys()) - set(include))
            exclude = set(exclude + exclude_attributes) if exclude else exclude_attributes
        if exclude:
            for exclude_attribute in exclude:
                dumped.pop(exclude_attribute, None)
        return dumped

    def create_object_from_schema(self, schema: SchemaMeta, data: dict, access: BaseWorld.Access):
        obj_schema = schema()
        obj = obj_schema.load(data)
        obj.access = self._get_allowed_from_access(access)
        return obj.store(self._data_svc.ram)

    async def create_on_disk_object(self, data: dict, access: dict, ram_key: str, id_property: str, obj_class: type):
        obj_id = data.get(id_property) or str(uuid.uuid4())
        data[id_property] = obj_id

        file_path = await self._get_new_object_file_path(data[id_property], ram_key)
        allowed = self._get_allowed_from_access(access)
        await self._save_and_reload_object(file_path, data, obj_class, allowed)
        return next(self.find_objects(ram_key, {id_property: obj_id}))