o2ims/bootstrap.py

   1 # Copyright (C) 2021 Wind River Systems, Inc.
   2 #
   3 #  Licensed under the Apache License, Version 2.0 (the "License");
   4 #  you may not use this file except in compliance with the License.
   5 #  You may obtain a copy of the License at
   6 #
   7 #      http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 #  Unless required by applicable law or agreed to in writing, software
  10 #  distributed under the License is distributed on an "AS IS" BASIS,
  11 #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 #  See the License for the specific language governing permissions and
  13 #  limitations under the License.
  14
  15 import inspect
  16 from typing import Callable
  17 from o2ims.adapter import orm, redis_eventpublisher
  18 from o2ims.adapter.notifications import AbstractNotifications,\
  19     SmoO2Notifications
  20
  21 from o2ims.service import handlers, messagebus, unit_of_work
  22
  23
  24 def bootstrap(
  25     start_orm: bool = True,
  26     uow: unit_of_work.AbstractUnitOfWork = unit_of_work.SqlAlchemyUnitOfWork(),
  27     notifications: AbstractNotifications = None,
  28     publish: Callable = redis_eventpublisher.publish,
  29 ) -> messagebus.MessageBus:
  30
  31     if notifications is None:
  32         notifications = SmoO2Notifications()
  33
  34     if start_orm:
  35         orm.start_o2ims_mappers()
  36
  37     dependencies = {"uow": uow, "notifications": notifications,
  38                     "publish": publish}
  39     injected_event_handlers = {
  40         event_type: [
  41             inject_dependencies(handler, dependencies)
  42             for handler in event_handlers
  43         ]
  44         for event_type, event_handlers in handlers.EVENT_HANDLERS.items()
  45     }
  46     injected_command_handlers = {
  47         command_type: inject_dependencies(handler, dependencies)
  48         for command_type, handler in handlers.COMMAND_HANDLERS.items()
  49     }
  50
  51     return messagebus.MessageBus(
  52         uow=uow,
  53         event_handlers=injected_event_handlers,
  54         command_handlers=injected_command_handlers,
  55     )
  56
  57
  58 def inject_dependencies(handler, dependencies):
  59     params = inspect.signature(handler).parameters
  60     deps = {
  61         name: dependency
  62         for name, dependency in dependencies.items()
  63         if name in params
  64     }
  65     return lambda message: handler(message, **deps)