68 lines
2.8 KiB
Python
68 lines
2.8 KiB
Python
import logging
|
|
from typing import Tuple
|
|
|
|
from sqlalchemy.orm.exc import NoResultFound
|
|
|
|
from app.extensions import db
|
|
from app.models.mirrors import Origin, StaticOrigin
|
|
from app.terraform import BaseAutomation
|
|
|
|
|
|
class StaticMetaAutomation(BaseAutomation):
|
|
short_name = "static_meta"
|
|
description = "Housekeeping for static origins"
|
|
frequency = 1
|
|
|
|
def automate(self, full: bool = False) -> Tuple[bool, str]:
|
|
"""
|
|
Create Origins for each StaticOrigin where an Origin does not already exist.
|
|
Set origin_domain_name of the StaticOrigin to the domain_name of the Origin.
|
|
Update Origin's auto_rotation attribute to match StaticOrigin's setting.
|
|
Remove Origins for StaticOrigins with non-null destroy value.
|
|
"""
|
|
|
|
# Step 1: Create Origins for StaticOrigins without existing Origins
|
|
static_origins = StaticOrigin.query.all()
|
|
for static_origin in static_origins:
|
|
if static_origin.origin_domain_name is not None:
|
|
try:
|
|
# Check if an Origin with the same domain name already exists
|
|
origin = Origin.query.filter_by(
|
|
domain_name=static_origin.origin_domain_name
|
|
).one()
|
|
# Keep auto rotation value in sync
|
|
origin.auto_rotation = static_origin.auto_rotate
|
|
except NoResultFound:
|
|
# Create a new Origin since it doesn't exist
|
|
origin = Origin(
|
|
group_id=static_origin.group_id,
|
|
description=f"PORTAL !! DO NOT DELETE !! Automatically created web origin for static origin "
|
|
f"#{static_origin.id}",
|
|
domain_name=static_origin.origin_domain_name,
|
|
auto_rotation=static_origin.auto_rotate,
|
|
smart=False,
|
|
assets=False,
|
|
)
|
|
db.session.add(origin)
|
|
logging.debug(
|
|
f"Created Origin with domain name {origin.domain_name}"
|
|
)
|
|
|
|
# Step 2: Remove Origins for StaticOrigins with non-null destroy value
|
|
static_origins_with_destroyed = StaticOrigin.query.filter(
|
|
StaticOrigin.destroyed.isnot(None)
|
|
).all()
|
|
for static_origin in static_origins_with_destroyed:
|
|
try:
|
|
origin = Origin.query.filter_by(
|
|
domain_name=static_origin.origin_domain_name,
|
|
destroyed=None,
|
|
).one()
|
|
origin.destroy()
|
|
logging.debug(f"Destroyed Origin with domain name {origin.domain_name}")
|
|
except NoResultFound:
|
|
continue
|
|
|
|
db.session.commit()
|
|
|
|
return True, ""
|