- def create_tables(self):
- try:
- db.connect()
- db.create_tables([ScalingGroup, ScalingPolicy, ScalingCriteria, ScalingAlarm])
- db.close()
- except Exception:
- log.exception("Error creating tables: ")
-
- def get_alarm(self, alarm_id: str):
- return ScalingAlarm.select().where(ScalingAlarm.alarm_id == alarm_id).get()
+ 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()