Skip to content

ofrak_context.py

ofrak.ofrak_context

OFRAK

__init__(self, logging_level=30, exclude_components_missing_dependencies=False) special

Set up the OFRAK environment that a script will use.

Parameters:

Name Type Description Default
logging_level int

Logging level of OFRAK instance (logging.DEBUG, logging.WARNING, etc.)

 30
exclude_components_missing_dependencies bool

When initializing OFRAK, check each component's dependency and do not use any components missing some dependencies

 False
Source code in ofrak/ofrak_context.py 
def __init__(
    self,
    logging_level: int = DEFAULT_LOG_LEVEL,
    exclude_components_missing_dependencies: bool = False,
):
    """
    Set up the OFRAK environment that a script will use.

    :param logging_level: Logging level of OFRAK instance (logging.DEBUG, logging.WARNING, etc.)
    :param exclude_components_missing_dependencies: When initializing OFRAK, check each component's dependency and do
    not use any components missing some dependencies
    """
    logging.basicConfig(level=logging_level, format="[%(filename)15s:%(lineno)5s] %(message)s")
    logging.getLogger().addHandler(logging.FileHandler(DEFAULT_OFRAK_LOG_FILE))
    logging.getLogger().setLevel(logging_level)
    logging.captureWarnings(True)
    self.injector = DependencyInjector()
    self._discovered_modules: List[ModuleType] = []
    self._exclude_components_missing_dependencies = exclude_components_missing_dependencies
    self._id_service: Optional[IDServiceInterface] = None

create_ofrak_context(self) async

Create the OFRAKContext and start all its services.

Source code in ofrak/ofrak_context.py 
async def create_ofrak_context(self) -> OFRAKContext:
    """
    Create the OFRAKContext and start all its services.
    """
    self._setup()
    component_locator = await self.injector.get_instance(ComponentLocatorInterface)

    resource_factory = await self.injector.get_instance(ResourceFactory)
    components = await self._get_discovered_components()
    component_locator.add_components(components, self._discovered_modules)

    id_service = await self.injector.get_instance(IDServiceInterface)
    data_service = await self.injector.get_instance(DataServiceInterface)
    resource_service = await self.injector.get_instance(ResourceServiceInterface)
    job_service = await self.injector.get_instance(JobServiceInterface)
    all_services = await self.injector.get_instance(List[AbstractOfrakService])

    ofrak_context = OFRAKContext(
        self.injector,
        resource_factory,
        component_locator,
        id_service,
        data_service,
        resource_service,
        job_service,
        all_services,
    )
    await ofrak_context.start_context()
    return ofrak_context

_setup(self) private

Discover common OFRAK services and components.

Source code in ofrak/ofrak_context.py 
def _setup(self):
    """Discover common OFRAK services and components."""
    import ofrak

    self.discover(ofrak)
    self.discover(ofrak_patch_maker)

    if self._id_service:
        self.injector.bind_instance(self._id_service)