shanbady/separate database router and schema for program certificates

main/routers.py

@@ -0,0 +1,13 @@
+from django.conf import settings
+
+
+class ReadOnlyModelError(Exception):
+    """Raised when a write was attempted on a read-only model"""
+
+
+class ExternalSchemaRouter:
+    def db_for_write(self, model, **meta):  # noqa: ARG002
+        model_name = model._meta.model_name  # noqa: SLF001
+        if model_name in settings.EXTERNAL_MODELS:
+            exception_message = f"model {model_name} is read only"
+            raise ReadOnlyModelError(exception_message)

main/routers_test.py

@@ -0,0 +1,12 @@
+import pytest
+
+from main.routers import ReadOnlyModelError
+from profiles.factories import ProgramCertificateFactory
+
+
+def test_external_tables_are_readonly():
+    """
+    Test that external tables cannot be written to
+    """
+    with pytest.raises(ReadOnlyModelError):
+        ProgramCertificateFactory(user_email="test@test.com", program_title="test")

main/settings.py

@@ -227,6 +227,10 @@
 
 DATABASES = {"default": DEFAULT_DATABASE_CONFIG}
 
+DATABASE_ROUTERS = ["main.routers.ExternalSchemaRouter"]
+
+EXTERNAL_MODELS = ["programcertificate"]
+
 # Internationalization
 # https://docs.djangoproject.com/en/1.8/topics/i18n/
 

profiles/migrations/0018_alter_programletter_certificate_and_more.py

@@ -0,0 +1,95 @@
+# Generated by Django 4.2.11 on 2024-03-14 23:16
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("profiles", "0017_programletter"),
+    ]
+
+    operations = [
+        migrations.RunSQL('DROP TABLE "external.programcertificate" CASCADE'),
+        migrations.CreateModel(
+            name="ProgramCertificate",
+            fields=[
+                (
+                    "id",
+                    models.AutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                ("user_edxorg_id", models.IntegerField(blank=True, null=True)),
+                ("micromasters_program_id", models.IntegerField(blank=True, null=True)),
+                ("mitxonline_program_id", models.IntegerField(blank=True, null=True)),
+                (
+                    "user_edxorg_username",
+                    models.CharField(blank=True, max_length=256),
+                ),
+                (
+                    "user_email",
+                    models.CharField(blank=False, max_length=256),
+                ),
+                (
+                    "program_title",
+                    models.CharField(blank=False, max_length=256),
+                ),
+                (
+                    "user_gender",
+                    models.CharField(blank=True, max_length=256),
+                ),
+                (
+                    "user_address_city",
+                    models.CharField(blank=True, max_length=256),
+                ),
+                (
+                    "user_first_name",
+                    models.CharField(blank=True, max_length=256),
+                ),
+                (
+                    "user_last_name",
+                    models.CharField(blank=True, max_length=256),
+                ),
+                (
+                    "user_full_name",
+                    models.CharField(blank=True, max_length=256),
+                ),
+                (
+                    "user_year_of_birth",
+                    models.CharField(blank=True, max_length=256),
+                ),
+                (
+                    "user_country",
+                    models.CharField(blank=True, max_length=256),
+                ),
+                (
+                    "user_address_postal_code",
+                    models.CharField(blank=True, max_length=256),
+                ),
+                (
+                    "user_street_address",
+                    models.CharField(blank=True, max_length=256),
+                ),
+                (
+                    "user_address_state_or_territory",
+                    models.CharField(blank=True, max_length=256),
+                ),
+                (
+                    "user_mitxonline_username",
+                    models.CharField(blank=True, max_length=256),
+                ),
+                (
+                    "program_completion_timestamp",
+                    models.DateTimeField(blank=True, null=True),
+                ),
+            ],
+            options={"db_table": '"external"."programcertificate"'},
+        ),
+        migrations.AlterModelTable(
+            name="programcertificate",
+            table='"external"."programcertificate"',
+        ),
+    ]

profiles/migrations/0019_alter_programcertificate_options_and_more.py

@@ -0,0 +1,26 @@
+# Generated by Django 4.2.11 on 2024-03-15 13:43
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("profiles", "0018_alter_programletter_certificate_and_more"),
+    ]
+
+    operations = [
+        migrations.AlterModelOptions(
+            name="programcertificate",
+            options={"managed": False},
+        ),
+        migrations.AlterField(
+            model_name="programletter",
+            name="certificate",
+            field=models.ForeignKey(
+                null=True,
+                on_delete=django.db.models.deletion.CASCADE,
+                to="profiles.programcertificate",
+            ),
+        ),
+    ]

profiles/models.py

@@ -179,7 +179,7 @@ class ProgramCertificate(models.Model):
 
     class Meta:
         managed = False
-        db_table = "external.programcertificate"
+        db_table = '"external"."programcertificate"'
 
     def __str__(self):
         return f"program certificate: {self.user_full_name} - {self.program_title}"
@@ -192,7 +192,9 @@ class ProgramLetter(models.Model):
 
     id = models.UUIDField(primary_key=True, default=uuid4, editable=False)
     user = models.ForeignKey(User, on_delete=models.CASCADE)
-    certificate = models.ForeignKey(ProgramCertificate, on_delete=models.CASCADE)
+    certificate = models.ForeignKey(
+        ProgramCertificate, on_delete=models.CASCADE, null=True
+    )
 
     def __str__(self):
         return (

profiles/models_test.py

@@ -65,21 +65,19 @@ def test_external_schema_exists():
     with connection.cursor() as cursor:
         cursor.execute(
             """
-            SELECT 1
-                FROM pg_catalog.pg_class
-                WHERE relname = 'external.programcertificate'
-                AND relkind = 'r';
+            SELECT schema_name FROM information_schema.schemata WHERE schema_name = 'external';
             """
         )
-        assert cursor.fetchone()[0] == 1
+        assert cursor.fetchone()[0] == "external"
 
 
 @pytest.mark.django_db()
-def test_program_letter_model_strings(user):
+def test_program_letter_model_strings(user, settings):
     """
     Test that ProgramCertificate and ProgramLetter string methods
     return what we expect
     """
+    settings.DATABASE_ROUTERS = []
     cert = ProgramCertificateFactory(
         user_full_name="test user", program_title="test program"
     )

profiles/utils_test.py

@@ -5,7 +5,6 @@
 from urllib.parse import parse_qs, urlparse
 
 import pytest
-from django.conf import settings
 from PIL import Image
 
 from main.factories import UserFactory
@@ -190,11 +189,14 @@ def test_generate_initials(text, initials):
 
 
 @pytest.mark.django_db()
-def test_fetch_program_letter_template_data_malformed_api_response(mocker, user):
+def test_fetch_program_letter_template_data_malformed_api_response(
+    mocker, user, settings
+):
     """
     Tests that a malformed response from micromasters api
     causes fetch_program_letter_template_data to return None
     """
+    settings.DATABASE_ROUTERS = []
     settings.MICROMASTERS_CMS_API_URL = "http://test.com"
     mm_api_response = mocker.Mock()
     mm_api_response.configure_mock(**{"json.return_value": {"some": "json"}})
@@ -205,11 +207,12 @@ def test_fetch_program_letter_template_data_malformed_api_response(mocker, user)
 
 
 @pytest.mark.django_db()
-def test_fetch_program_letter_template_data_has_results(mocker, user):
+def test_fetch_program_letter_template_data_has_results(mocker, user, settings):
     """
     Tests that a response from micromasters api
     with a result returns properly
     """
+    settings.DATABASE_ROUTERS = []
     settings.MICROMASTERS_CMS_API_URL = "http://test.com"
     expected_item = {"test": "test"}
     mm_api_response = mocker.Mock()

profiles/views_test.py

@@ -330,14 +330,14 @@ def test_get_user_by_me(mocker, client, user, is_anonymous):
 
 @pytest.mark.parametrize("is_anonymous", [True, False])
 def test_letter_intercept_view_generates_program_letter(
-    mocker, client, user, is_anonymous
+    mocker, client, user, is_anonymous, settings
 ):
     """
     Test that the letter intercept view generates a
     ProgramLetter and then passes the user along to the display.
     Also test that anonymous users do not generate letters and cant access this page
     """
-
+    settings.DATABASE_ROUTERS = []
     micromasters_program_id = 1
     if not is_anonymous:
         client.force_login(user)
@@ -364,11 +364,12 @@ def test_letter_intercept_view_generates_program_letter(
 
 
 @pytest.mark.parametrize("is_anonymous", [True, False])
-def test_program_letter_api_view(mocker, client, user, is_anonymous):
+def test_program_letter_api_view(mocker, client, user, is_anonymous, settings):
     """
     Test that the program letter display page is viewable by
     all users logged in or not
     """
+    settings.DATABASE_ROUTERS = []
     mock_return_value = {
         "id": 4,
         "meta": {},