Fix crash on missing sequences for new schema permissions apply

CHANGES.md

@@ -1,3 +1,7 @@
+# 2024-10-01 (6.1.1)
+
+* Fixed crash in `schema permissions apply` for tables that don't have a sequence in the database.
+
 # 2024-10-01 (6.1)
 
 * Added SQL SEQUENCE permissions in `schema permissions apply` to make `pg_dump` access easier.

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = amsterdam-schema-tools
-version = 6.1
+version = 6.1.1
 url = https://github.com/amsterdam/schema-tools
 license = Mozilla Public 2.0
 author = Team Data Diensten, van het Dataplatform onder de Directie Digitale Voorzieningen (Gemeente Amsterdam)

src/schematools/permissions/db.py

@@ -193,9 +193,10 @@ def set_dataset_write_permissions(
             echo=echo,
             dry_run=dry_run,
         )
-        if table.is_autoincrement:
+        if table.is_autoincrement and (
+            sequence_name := _get_sequence_name(session, table_name, "id")
+        ):
             # Get PostgreSQL generated sequence for 'id' column that Django added.
-            sequence_name = _get_sequence_name(session, table_name, "id")
             _execute_grant(
                 session,
                 grant(
@@ -312,9 +313,12 @@ def _fetch_grantees(scopes: frozenset[str]) -> list[str]:
                         grantees=grantees,
                     )
                 )
-                if field.is_primary and table.is_autoincrement:
-                    # Get PostgreSQL generated sequence for 'id' column that Django added.
-                    sequence_name = _get_sequence_name(session, table_name, "id")
+                # Get PostgreSQL generated sequence for 'id' column that Django added.
+                if (
+                    field.is_primary
+                    and table.is_autoincrement
+                    and (sequence_name := _get_sequence_name(session, table_name, "id"))
+                ):
                     all_scopes.append(
                         GrantParam(
                             privileges=["SELECT"],
@@ -334,8 +338,9 @@ def _fetch_grantees(scopes: frozenset[str]) -> list[str]:
                         grantees=grantees,
                     )
                 )
-                if table.is_autoincrement:
-                    sequence_name = _get_sequence_name(session, table_name, "id")
+                if table.is_autoincrement and (
+                    sequence_name := _get_sequence_name(session, table_name, "id")
+                ):
                     all_scopes.append(
                         GrantParam(
                             privileges=["SELECT"],
@@ -581,8 +586,9 @@ def _revoke_all_privileges_from_read_and_write_roles(
                 revoke_statements.append(
                     f"REVOKE ALL PRIVILEGES ON {pg_schema}.{table.db_name} FROM {rolname[0]}"
                 )
-                if table.is_autoincrement:
-                    sequence_name = _get_sequence_name(session, table.db_name, "id")
+                if table.is_autoincrement and (
+                    sequence_name := _get_sequence_name(session, table.db_name, "id")
+                ):
                     revoke_statements.append(
                         f"REVOKE ALL PRIVILEGES ON SEQUENCE {pg_schema}.{sequence_name}"
                         f" FROM {rolname[0]}"
@@ -673,7 +679,13 @@ def _get_sequence_name(session: Session, table_name: str, column: str) -> str |
             text("SELECT pg_get_serial_sequence(:table, :column)"),
             {"table": table_name, "column": column},
         ).first()
-        value = row[0].replace("public.", "").replace('"', "") if row is not None else None
+        value = (
+            row[0].replace("public.", "").replace('"', "")
+            if row is not None and row[0] is not None
+            else None
+        )
+        if not value:
+            logger.debug("No sequence found for %s.%s", table_name, column)
         existing_sequences[key] = value
         return value