ofrak_context.py

ofrak.ofrak_context

OFRAK

__init__(self, logging_level=30, log_file=None, exclude_components_missing_dependencies=False, verify_license=True) 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
verify_license	bool	Verify OFRAK license	True

Source code in ofrak/ofrak_context.py

def __init__(
    self,
    logging_level: int = DEFAULT_LOG_LEVEL,
    log_file: Optional[str] = None,
    exclude_components_missing_dependencies: bool = False,
    verify_license: bool = True,
):
    """
    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
    :param verify_license: Verify OFRAK license
    """
    if verify_license:
        verify_registered_license()
    logging.basicConfig(level=logging_level, format="[%(filename)15s:%(lineno)5s] %(message)s")
    if log_file is None:
        log_file = DEFAULT_LOG_FILE
    logging.getLogger().addHandler(logging.FileHandler(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)