new app versioning

src/country_workspace/config/settings.py

@@ -44,7 +44,7 @@
     "country_workspace.security",
     "country_workspace.apps.Config",
     "country_workspace.workspaces.apps.Config",
-    "country_workspace.versions",
+    "country_workspace.versioning",
 )
 
 MIDDLEWARE = (

src/country_workspace/migrations/0001_initial.py

@@ -1,4 +1,4 @@
-# Generated by Django 5.1.1 on 2024-10-14 10:37
+# Generated by Django 5.1.1 on 2024-10-15 16:46
 
 import django.contrib.auth.models
 import django.contrib.auth.validators

src/country_workspace/sync/office.py

@@ -62,21 +62,29 @@ def sync_programs(limit_to_office: "Optional[Office]" = None) -> int:
 
 def sync_lookup(sl: SyncLog):
     fd = sl.content_object
+    if not fd:
+        return
     client = HopeClient()
     record = client.get_lookup(sl.data["remote_url"])
     choices = []
     for k, v in record.items():
         choices.append((k, v))
-
+    if not fd.attrs:
+        fd.attrs = {}
     fd.attrs["choices"] = choices
     fd.save()
     sl.last_update_date = timezone.now()
     sl.save()
 
 
+def sync_lookups() -> bool:
+    for sl in SyncLog.objects.filter(object_id__gt=0).exclude(content_type__isnull=True):
+        sync_lookup(sl)
+    return True
+
+
 def sync_all() -> bool:
     sync_offices()
     sync_programs()
-    for sl in SyncLog.objects.filter(object_id__gt=0):
-        sync_lookup(sl)
+    sync_lookups()
     return True

src/country_workspace/versioning/api.py

@@ -0,0 +1,6 @@
+from country_workspace.versioning.management.manager import Manager
+
+
+def run_scripts():
+    m = Manager()
+    m.forward(m.max_version)

src/country_workspace/versions/utils.py → src/country_workspace/versioning/checkers.py

@@ -1,51 +1,42 @@
-from typing import TYPE_CHECKING
-
 from django import forms
 from django.conf import settings
 from django.utils.text import slugify
 
-from country_workspace.constants import HOUSEHOLD_CHECKER_NAME, INDIVIDUAL_CHECKER_NAME
-
-if TYPE_CHECKING:
-    from hope_flex_fields.models import DataChecker, FieldDefinition, Fieldset
+from hope_flex_fields.models import DataChecker, FieldDefinition, Fieldset
 
+from country_workspace.constants import HOUSEHOLD_CHECKER_NAME, INDIVIDUAL_CHECKER_NAME
 
-def create_hope_field_definitions(apps, schema_editor):
-    fd: "FieldDefinition" = apps.get_model("hope_flex_fields", "FieldDefinition")
 
+def create_hope_field_definitions():
     for m in settings.LOOKUPS:
         n = f"HOPE HH {m}"
-        fd.objects.get_or_create(name=n, slug=slugify(n), field_type=forms.ChoiceField)
-    fd.objects.get_or_create(
+        FieldDefinition.objects.get_or_create(name=n, slug=slugify(n), field_type=forms.ChoiceField)
+    FieldDefinition.objects.get_or_create(
         name="HOPE IND Gender",
         slug=slugify("HOPE IND Gender"),
         attrs={"choices": [["FEMALE", "FEMALE"], ["MALE", "MALE"], ["UNKNOWN", "UNKNOWN"]]},
         field_type=forms.ChoiceField,
     )
-    fd.objects.get_or_create(
+    FieldDefinition.objects.get_or_create(
         name="HOPE IND Disability",
         slug=slugify("HOPE IND Disability"),
         field_type=forms.ChoiceField,
         attrs={"choices": [["not disabled", "not disabled"], ["disabled", "disabled"]]},
     )
 
 
-def create_hope_core_fieldset(apps, schema_editor):
-    dc: "DataChecker" = apps.get_model("hope_flex_fields", "DataChecker")
-    fs: "Fieldset" = apps.get_model("hope_flex_fields", "Fieldset")
-    fd: "FieldDefinition" = apps.get_model("hope_flex_fields", "FieldDefinition")
-
-    _char = fd.objects.get(field_type=forms.CharField)
-    _date = fd.objects.get(field_type=forms.DateField)
-    _bool = fd.objects.get(field_type=forms.BooleanField)
-    _int = fd.objects.get(field_type=forms.IntegerField)
+def create_hope_core_fieldset():
+    _char = FieldDefinition.objects.get(field_type=forms.CharField)
+    _date = FieldDefinition.objects.get(field_type=forms.DateField)
+    _bool = FieldDefinition.objects.get(field_type=forms.BooleanField)
+    _int = FieldDefinition.objects.get(field_type=forms.IntegerField)
 
-    _h_relationship = fd.objects.get(slug="hope-hh-relationship")
-    _h_residence = fd.objects.get(slug="hope-hh-residencestatus")
-    _i_gender = fd.objects.get(slug="hope-ind-gender")
-    _i_disability = fd.objects.get(slug="hope-ind-disability")
+    _h_relationship = FieldDefinition.objects.get(slug="hope-hh-relationship")
+    _h_residence = FieldDefinition.objects.get(slug="hope-hh-residencestatus")
+    _i_gender = FieldDefinition.objects.get(slug="hope-ind-gender")
+    _i_disability = FieldDefinition.objects.get(slug="hope-ind-disability")
 
-    hh_fs, __ = fs.objects.get_or_create(name=HOUSEHOLD_CHECKER_NAME)
+    hh_fs, __ = Fieldset.objects.get_or_create(name=HOUSEHOLD_CHECKER_NAME)
     hh_fs.fields.get_or_create(
         name="address",
         attrs={"label": "Household ID", "required": True},
@@ -144,7 +135,7 @@ def create_hope_core_fieldset(apps, schema_editor):
     # hh_fs.fields.get_or_create(field=_bf, name="hh_latrine_h_f", attrs={"label": "Latrine"})
     # hh_fs.fields.get_or_create(field=_bf, name="hh_electricity_h_f")
 
-    ind_fs, __ = fs.objects.get_or_create(name="HOPE individual core")
+    ind_fs, __ = Fieldset.objects.get_or_create(name="HOPE individual core")
     ind_fs.fields.get_or_create(
         name="address",
         attrs={"label": "Household ID", "required": True},
@@ -228,7 +219,7 @@ def create_hope_core_fieldset(apps, schema_editor):
         field=_h_relationship,
     )
 
-    hh_dc, __ = dc.objects.get_or_create(name=HOUSEHOLD_CHECKER_NAME)
+    hh_dc, __ = DataChecker.objects.get_or_create(name=HOUSEHOLD_CHECKER_NAME)
     hh_dc.fieldsets.add(hh_fs)
-    ind_dc, __ = dc.objects.get_or_create(name=INDIVIDUAL_CHECKER_NAME)
+    ind_dc, __ = DataChecker.objects.get_or_create(name=INDIVIDUAL_CHECKER_NAME)
     ind_dc.fieldsets.add(ind_fs)

src/country_workspace/versioning/management/commands/applyversion.py

@@ -0,0 +1,26 @@
+from django.core.management.base import BaseCommand, no_translations
+
+from country_workspace.versioning.management.manager import Manager
+
+
+class Command(BaseCommand):
+    help = "Creates new version for apps."
+
+    def add_arguments(self, parser):
+        parser.add_argument("num", nargs="?", help="Specify the version label")
+
+    @no_translations
+    def handle(self, num, **options):
+        m = Manager()
+        if not num:
+            num = m.max_version
+        print(f"Available update {m.max_version}")
+        print(f"Applied update {m.max_applied_version}")
+        if num == "zero":
+            m.zero()
+        else:
+            num = int(num)
+            if not num:
+                num = m.max_applied_version
+            if num >= m.max_applied_version:
+                m.forward(num)

src/country_workspace/versioning/management/commands/makeversion.py

@@ -0,0 +1,48 @@
+import re
+from pathlib import Path
+
+from django.core.management.base import BaseCommand, no_translations
+from django.utils.timezone import now
+
+import country_workspace
+
+VERSION_TEMPLATE = """# Generated by HCW %(version)s on %(timestamp)s
+
+class Version:
+    operations = []
+"""
+
+regex = re.compile(r"(\d+).*")
+
+
+def get_version(filename):
+    if m := regex.match(filename):
+        return int(m.group(1))
+    return None
+
+
+ts = now().strftime("%Y_%m_%d_%H%M%S")
+
+
+class Command(BaseCommand):
+    help = "Creates new version"
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "label",
+            nargs="?",
+            help="Specify the version label",
+        )
+
+    @no_translations
+    def handle(self, label, **options):
+        folder = Path(__file__).parent.parent.parent / "versions"
+        last_ver = 0
+        for filename in folder.iterdir():
+            if ver := get_version(filename.name):
+                last_ver = max(last_ver, ver)
+        new_ver = last_ver + 1
+        dest_file = folder / "{:>04}_{}.py".format(new_ver, label or ts)
+        with dest_file.open("w") as f:
+            f.write(VERSION_TEMPLATE % {"timestamp": ts, "version": country_workspace.VERSION})
+        print(f"Created version {dest_file.name}")

src/country_workspace/versioning/management/commands/showversion.py

@@ -0,0 +1,39 @@
+import re
+from pathlib import Path
+
+from django.core.management.base import BaseCommand, no_translations
+
+from country_workspace.versioning.models import Version
+
+regex = re.compile(r"(\d+).*")
+
+
+def get_version(filename):
+    if m := regex.match(filename):
+        return int(m.group(1))
+    return None
+
+
+class Command(BaseCommand):
+    help = "Creates new version for apps."
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "num",
+            nargs="?",
+            help="Specify the version label",
+        )
+
+    @no_translations
+    def handle(self, *app_labels, **options):
+        folder = Path(__file__).parent.parent.parent / "versions"
+        existing = {}
+        applied = list(Version.objects.order_by("name").values_list("name", flat=True))
+        for filename in sorted(folder.iterdir()):
+            if ver := get_version(filename.name):
+                existing[ver] = filename.name
+        for filename in existing.values():
+            if filename in applied:
+                print(f"[x] {filename}")
+            else:
+                print(f"[ ] {filename}")

src/country_workspace/versioning/management/manager.py

@@ -0,0 +1,89 @@
+import importlib.util
+import re
+from pathlib import Path
+from typing import Callable
+
+from country_workspace import VERSION
+from country_workspace.versioning.models import Version
+
+regex = re.compile(r"(\d+).*")
+default_folder = Path(__file__).parent.parent / "scripts"
+
+
+def get_version(filename):
+    if m := regex.match(filename):
+        return int(m.group(1))
+    return None
+
+
+def get_funcs(filename: Path, direction: str = "forward"):
+    if not filename.exists():  # pragma: no cover
+        raise FileNotFoundError(filename)
+    spec = importlib.util.spec_from_file_location("version", filename.absolute())
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    funcs = []
+    for op in module.Version.operations:
+        if isinstance(op, (list, tuple)):
+            if direction == "forward":
+                funcs.append(op[0])
+            else:
+                funcs.append(op[1])
+        else:
+            if direction == "forward":
+                funcs.append(op)
+            else:
+                funcs.append(lambda: True)
+
+    return funcs
+
+
+class Manager:
+    def __init__(self, folder: Path = default_folder):
+        self.folder = folder
+        self.existing = []
+        self.applied = list(Version.objects.order_by("name").values_list("name", flat=True))
+        self.max_version = 0
+        self.max_applied_version = 0
+        for applied in self.applied:
+            self.max_applied_version = max(get_version(applied), self.max_applied_version)
+
+        for filename in sorted(self.folder.iterdir()):
+            if v := get_version(filename.name):
+                self.existing.append(filename)
+                self.max_version = max(self.max_version, v)
+
+    def zero(self):
+        self.backward(0)
+
+    def forward(self, to_num) -> list[tuple[Path, list[Callable[[None], None]]]]:
+        print("Upgrading...")
+        processed = []
+        for entry in self.existing:
+            if get_version(entry.stem) > to_num:
+                break
+            if entry.name not in self.applied:
+                funcs = get_funcs(entry, direction="forward")
+                print(f"    Applying {entry.stem}")
+                for func in funcs:
+                    func()
+                Version.objects.create(name=entry.name, version=VERSION)
+                processed.append((entry, funcs))
+        self.applied = list(Version.objects.order_by("name").values_list("name", flat=True))
+        return processed
+
+    def backward(self, to_num) -> list[tuple[Path, list[Callable[[None], None]]]]:
+        print("Downgrading...")
+        processed = []
+        for entry in reversed(self.applied):
+            if get_version(entry) <= to_num:
+                break
+            file_path = Path(self.folder) / entry
+            funcs = get_funcs(file_path, direction="backward")
+            print(f"  Discharging {file_path.stem}")
+            for func in funcs:
+                func()
+            Version.objects.get(name=file_path.name).delete()
+            processed.append((entry, funcs))
+        self.applied = list(Version.objects.order_by("name").values_list("name", flat=True))
+        return processed

src/country_workspace/versioning/migrations/0001_initial.py

@@ -0,0 +1,23 @@
+# Generated by Django 5.1.1 on 2024-10-15 16:45
+
+import django.utils.timezone
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = []
+
+    operations = [
+        migrations.CreateModel(
+            name="Version",
+            fields=[
+                ("id", models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")),
+                ("name", models.CharField(max_length=255, unique=True)),
+                ("version", models.CharField(max_length=255)),
+                ("applied", models.DateTimeField(default=django.utils.timezone.now)),
+            ],
+        ),
+    ]

src/country_workspace/versioning/models.py

@@ -0,0 +1,8 @@
+from django.db import models
+from django.utils.timezone import now
+
+
+class Version(models.Model):
+    name = models.CharField(max_length=255, unique=True)
+    version = models.CharField(max_length=255)
+    applied = models.DateTimeField(default=now)