o2app/adapter/unit_of_work.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 # pylint: disable=attribute-defined-outside-init
  16 from __future__ import annotations
  17 from sqlalchemy import create_engine
  18 from sqlalchemy.orm import sessionmaker
  19 from sqlalchemy.orm.session import Session
  20
  21 from o2common.config import config
  22 from o2common.service.unit_of_work import AbstractUnitOfWork
  23
  24 from o2ims.adapter import ocloud_repository, alarm_repository, alarm_loader
  25 from o2dms.adapter import dms_repository
  26
  27 from o2common.helper import o2logging
  28 logger = o2logging.get_logger(__name__)
  29
  30
  31 engine = create_engine(
  32     config.get_postgres_uri(),
  33     isolation_level="REPEATABLE READ",
  34     pool_size=200, max_overflow=0,
  35     pool_recycle=3600
  36 )
  37
  38 DEFAULT_SESSION_FACTORY = sessionmaker(
  39     autocommit=False, autoflush=False, bind=engine
  40 )
  41
  42
  43 class SqlAlchemyUnitOfWork(AbstractUnitOfWork):
  44
  45     def __init__(self, session_factory=DEFAULT_SESSION_FACTORY):
  46         self.session_factory = session_factory
  47
  48     def __enter__(self):
  49         self.session = self.session_factory()  # type: Session
  50         self.oclouds = ocloud_repository\
  51             .OcloudSqlAlchemyRepository(self.session)
  52         self.resource_types = ocloud_repository\
  53             .ResouceTypeSqlAlchemyRepository(self.session)
  54         self.resource_pools = ocloud_repository\
  55             .ResourcePoolSqlAlchemyRepository(self.session)
  56         self.resources = ocloud_repository\
  57             .ResourceSqlAlchemyRepository(self.session)
  58         self.subscriptions = ocloud_repository\
  59             .SubscriptionSqlAlchemyRepository(self.session)
  60         self.deployment_managers = ocloud_repository\
  61             .DeploymentManagerSqlAlchemyRepository(self.session)
  62         self.nfdeployment_descs = dms_repository\
  63             .NfDeploymentDescSqlAlchemyRepository(self.session)
  64         self.nfdeployments = dms_repository\
  65             .NfDeploymentSqlAlchemyRepository(self.session)
  66         self.ocloudvresources = dms_repository\
  67             .NfOCloudVResourceSqlAlchemyRepository(self.session)
  68         self.alarm_event_records = alarm_repository\
  69             .AlarmEventRecordSqlAlchemyRepository(self.session)
  70         self.alarm_definitions = alarm_repository\
  71             .AlarmDefinitionSqlAlchemyRepository(self.session)
  72         self.alarm_subscriptions = alarm_repository\
  73             .AlarmSubscriptionSqlAlchemyRepository(self.session)
  74         self.alarm_probable_causes = alarm_repository\
  75             .AlarmProbableCauseSqlAlchemyRepository(self.session)
  76
  77         # config file
  78         self.alarm_dictionaries = alarm_loader\
  79             .AlarmDictionaryConfigFileRepository()
  80         return super().__enter__()
  81
  82     def __exit__(self, *args):
  83         super().__exit__(*args)
  84         self.session.close()
  85
  86     def _commit(self):
  87         self.session.commit()
  88
  89     def rollback(self):
  90         self.session.rollback()
  91
  92     def _collect_new_events(self):
  93         for entry in self.oclouds.seen:
  94             # while hasattr(entry, 'events') and len(entry.events) > 0:
  95             while entry.events is not None and len(entry.events) > 0:
  96                 yield entry.events.pop(0)
  97         for entry in self.resource_pools.seen:
  98             while entry.events is not None and len(entry.events) > 0:
  99                 yield entry.events.pop(0)
 100         for entry in self.resources.seen:
 101             while entry.events is not None and len(entry.events) > 0:
 102                 yield entry.events.pop(0)
 103         for entry in self.resource_types.seen:
 104             while hasattr(entry, 'events') and len(entry.events) > 0:
 105                 yield entry.events.pop(0)
 106         for entry in self.deployment_managers.seen:
 107             while hasattr(entry, 'events') and len(entry.events) > 0:
 108                 yield entry.events.pop(0)
 109         for entry in self.subscriptions.seen:
 110             while hasattr(entry, 'events') and len(entry.events) > 0:
 111                 yield entry.events.pop(0)
 112         for entry in self.nfdeployment_descs.seen:
 113             while hasattr(entry, 'events') and len(entry.events) > 0:
 114                 yield entry.events.pop(0)
 115         for entry in self.nfdeployments.seen:
 116             while hasattr(entry, 'events') and len(entry.events) > 0:
 117                 yield entry.events.pop(0)
 118         for entry in self.ocloudvresources.seen:
 119             while hasattr(entry, 'events') and len(entry.events) > 0:
 120                 yield entry.events.pop(0)
 121         for entry in self.alarm_event_records.seen:
 122             while hasattr(entry, 'events') and len(entry.events) > 0:
 123                 yield entry.events.pop(0)
 124         for entry in self.alarm_subscriptions.seen:
 125             while hasattr(entry, 'events') and len(entry.events) > 0:
 126                 yield entry.events.pop(0)