+ def __init__(self, config: Config):
+ db.initialize(connect(config.get('sql', 'database_uri')))
+
+ def create_tables(self) -> None:
+ db.connect()
+ with db.atomic():
+ router = Router(db, os.path.dirname(migrations.__file__))
+ router.run()
+ db.close()
+
+
+class ScalingAlarmRepository:
+
+ @staticmethod
+ def list(*expressions) -> Iterable[ScalingAlarm]:
+ return ScalingAlarm.select().where(*expressions)
+
+ @staticmethod
+ def get(*expressions, join_classes: List = None) -> ScalingAlarm:
+ query = ScalingAlarm.select()
+ if join_classes:
+ for join_class in join_classes:
+ query = query.join(join_class)
+ return query.where(*expressions).get()
+
+ @staticmethod
+ def create(**query) -> ScalingAlarm:
+ return ScalingAlarm.create(**query)
+
+
+class ScalingGroupRepository:
+
+ @staticmethod
+ def list(*expressions) -> Iterable[ScalingGroup]:
+ return ScalingGroup.select().where(*expressions)
+
+ @staticmethod
+ def get(*expressions) -> ScalingGroup:
+ return ScalingGroup.select().where(*expressions).get()
+
+ @staticmethod
+ def create(**query) -> ScalingGroup:
+ return ScalingGroup.create(**query)
+
+
+class ScalingPolicyRepository:
+
+ @staticmethod
+ def list(*expressions, join_classes: List = None) -> Iterable[ScalingPolicy]:
+ query = ScalingPolicy.select()
+ if join_classes:
+ for join_class in join_classes:
+ query = query.join(join_class)
+ return query.where(*expressions)
+
+ @staticmethod
+ def get(*expressions, join_classes: List = None) -> ScalingPolicy:
+ query = ScalingPolicy.select()
+ if join_classes:
+ for join_class in join_classes:
+ query = query.join(join_class)
+ return query.where(*expressions).get()
+
+ @staticmethod
+ def create(**query) -> ScalingPolicy:
+ return ScalingPolicy.create(**query)
+
+
+class ScalingCriteriaRepository:
+
+ @staticmethod
+ def list(*expressions, join_classes: List = None) -> Iterable[ScalingCriteria]:
+ query = ScalingCriteria.select()
+ if join_classes:
+ for join_class in join_classes:
+ query = query.join(join_class)
+ return query.where(*expressions)
+
+ @staticmethod
+ def get(*expressions, join_classes: List = None) -> ScalingCriteria:
+ query = ScalingCriteria.select()
+ if join_classes:
+ for join_class in join_classes:
+ query = query.join(join_class)
+ return query.where(*expressions).get()
+
+ @staticmethod
+ def create(**query) -> ScalingCriteria:
+ return ScalingCriteria.create(**query)
+
+
+class VnfAlarmRepository:
+
+ @staticmethod
+ def list(*expressions) -> Iterable[VnfAlarm]:
+ return VnfAlarm.select().where(*expressions)
+
+ @staticmethod
+ def get(*expressions) -> VnfAlarm:
+ return VnfAlarm.select().where(*expressions).get()
+
+ @staticmethod
+ def create(**query) -> VnfAlarm:
+ return VnfAlarm.create(**query)
+
+
+class AlarmActionRepository:
+
+ @staticmethod
+ def list(*expressions) -> Iterable[AlarmAction]:
+ return AlarmAction.select().where(*expressions)
+
+ @staticmethod
+ def get(*expressions) -> AlarmAction:
+ return AlarmAction.select().where(*expressions).get()
+
+ @staticmethod
+ def create(**query) -> AlarmAction:
+ return AlarmAction.create(**query)