56 lines
1.4 KiB
Python
56 lines
1.4 KiB
Python
"""remove derived dkim dns artifacts
|
|
|
|
Revision ID: 20260520_0002
|
|
Revises: 20260520_0001
|
|
Create Date: 2026-05-20 00:00:00.000000+00:00
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from alembic import op
|
|
import sqlalchemy as sa
|
|
|
|
revision = "20260520_0002"
|
|
down_revision = "20260520_0001"
|
|
branch_labels = None
|
|
depends_on = None
|
|
|
|
|
|
def _is_dkim_lookup_artifact(value: object) -> bool:
|
|
return isinstance(value, str) and (
|
|
value.startswith("DKIM lookup failed")
|
|
or value.startswith("DKIM record not found")
|
|
)
|
|
|
|
|
|
def upgrade() -> None:
|
|
bind = op.get_bind()
|
|
snapshots = sa.table(
|
|
"domain_dns_snapshots",
|
|
sa.column("id", sa.Integer()),
|
|
sa.column("dkim_records_json", sa.Text()),
|
|
sa.column("errors_json", sa.Text()),
|
|
)
|
|
rows = bind.execute(sa.select(snapshots.c.id, snapshots.c.errors_json)).all()
|
|
for snapshot_id, errors_json in rows:
|
|
try:
|
|
errors = json.loads(errors_json or "[]")
|
|
except json.JSONDecodeError:
|
|
errors = []
|
|
if not isinstance(errors, list):
|
|
errors = []
|
|
cleaned_errors = [item for item in errors if not _is_dkim_lookup_artifact(item)]
|
|
bind.execute(
|
|
snapshots.update()
|
|
.where(snapshots.c.id == snapshot_id)
|
|
.values(
|
|
dkim_records_json="[]",
|
|
errors_json=json.dumps(cleaned_errors, sort_keys=True),
|
|
)
|
|
)
|
|
|
|
|
|
def downgrade() -> None:
|
|
pass
|