Add code-defined database queries for complex prefill sources

- Add database_query_key field to PrefillSource model
- Add execute_custom_query method to DatabaseDataSource
- Support dbquery.* prefix in prefill sources
- Queries defined in settings.FORMS_WORKFLOWS_DATABASE_QUERIES
- Enables complex SQL (JOINs, multiple WHERE conditions) without SQL injection concerns
- Add migration 0013_add_database_query_key
- Bump version to 0.7.7

Author: matteius

django_forms_workflows/__init__.py

@@ -3,7 +3,7 @@
 Enterprise-grade, database-driven form builder with approval workflows
 """
 
-__version__ = "0.7.6"
+__version__ = "0.7.7"
 __author__ = "Django Forms Workflows Contributors"
 __license__ = "LGPL-3.0-only"
 

django_forms_workflows/data_sources/database_source.py

@@ -384,6 +384,80 @@ def get_template_value(
             logger.error(f"Error getting template value: {e}")
             return None
 
+    def execute_custom_query(self, user, query_key: str) -> str | None:
+        """
+        Execute a code-defined database query.
+
+        Queries are defined in settings.FORMS_WORKFLOWS_DATABASE_QUERIES
+        and referenced by key. This allows complex queries (JOINs, multiple
+        WHERE conditions) while keeping SQL in version-controlled code.
+
+        Args:
+            user: Django User object
+            query_key: Key from FORMS_WORKFLOWS_DATABASE_QUERIES setting
+
+        Returns:
+            Query result (first column of first row) or None
+        """
+        if not user or not user.is_authenticated:
+            return None
+
+        try:
+            # Get query definition from settings
+            queries = getattr(settings, "FORMS_WORKFLOWS_DATABASE_QUERIES", {})
+            if query_key not in queries:
+                logger.error(
+                    f"Database query key '{query_key}' not found in "
+                    "FORMS_WORKFLOWS_DATABASE_QUERIES setting"
+                )
+                return None
+
+            query_config = queries[query_key]
+
+            # Extract configuration
+            query_text = query_config.get("query")
+            db_alias = query_config.get("db_alias")
+            user_field = query_config.get("user_field", "employee_id")
+
+            if not query_text or not db_alias:
+                logger.error(
+                    f"Database query '{query_key}' missing required 'query' or 'db_alias'"
+                )
+                return None
+
+            # Get user's lookup value
+            user_id = self._get_user_id(user, user_field)
+            if not user_id:
+                logger.debug(f"User {user.username} has no {user_field}")
+                return None
+
+            # Check database exists
+            databases = getattr(settings, "DATABASES", {})
+            if db_alias not in databases:
+                logger.error(f"Database alias '{db_alias}' not configured")
+                return None
+
+            # Execute the query
+            with connections[db_alias].cursor() as cursor:
+                cursor.execute(query_text, [user_id])
+                row = cursor.fetchone()
+
+                if row:
+                    value = row[0]
+                    # Strip whitespace for fixed-width columns
+                    if isinstance(value, str):
+                        value = value.strip()
+                    return value
+
+                logger.debug(
+                    f"No data found for user {user_id} with query '{query_key}'"
+                )
+                return None
+
+        except Exception as e:
+            logger.error(f"Error executing custom query '{query_key}': {e}")
+            return None
+
     def is_available(self) -> bool:
         """
         Check if database source is configured.

django_forms_workflows/forms.py

@@ -300,7 +300,8 @@ def _get_prefill_value(self, prefill_source, prefill_config=None):
         Supports:
         - user.* - User model fields
         - ldap.* - LDAP attributes
-        - db.* or {{ db.* }} - Database queries
+        - dbquery.* - Code-defined database queries (for complex SQL)
+        - db.* or {{ db.* }} - Simple database queries
         - api.* - API calls
         - current_date, current_datetime - Current date/time
         - last_submission - Previous submission data
@@ -325,6 +326,12 @@ def _get_prefill_value(self, prefill_source, prefill_config=None):
                 field_name = prefill_source.replace("ldap.", "")
                 return source.get_value(self.user, field_name) or ""
 
+            # Handle dbquery.* sources (code-defined complex queries)
+            elif prefill_source.startswith("dbquery."):
+                source = DatabaseDataSource()
+                query_key = prefill_source.replace("dbquery.", "")
+                return source.execute_custom_query(self.user, query_key) or ""
+
             # Handle db.* or {{ db.* }} sources
             elif prefill_source.startswith("db.") or prefill_source.startswith("{{"):
                 source = DatabaseDataSource()

django_forms_workflows/migrations/0013_add_database_query_key.py

@@ -0,0 +1,22 @@
+# Generated by Django 5.2.7 on 2026-01-22 00:08
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("django_forms_workflows", "0012_add_approval_step_fields"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="prefillsource",
+            name="database_query_key",
+            field=models.CharField(
+                blank=True,
+                help_text="Key from FORMS_WORKFLOWS_DATABASE_QUERIES setting. Use this for complex queries (JOINs, additional WHERE conditions). Takes precedence over db_schema/db_table/db_column fields.",
+                max_length=100,
+            ),
+        ),
+    ]

django_forms_workflows/models.py

@@ -201,6 +201,15 @@ class PrefillSource(models.Model):
         blank=True, null=True, help_text="Additional configuration as JSON"
     )
 
+    # Code-defined database query (for complex queries like JOINs)
+    database_query_key = models.CharField(
+        max_length=100,
+        blank=True,
+        help_text="Key from FORMS_WORKFLOWS_DATABASE_QUERIES setting. "
+        "Use this for complex queries (JOINs, additional WHERE conditions). "
+        "Takes precedence over db_schema/db_table/db_column fields.",
+    )
+
     # Metadata
     created_at = models.DateTimeField(auto_now_add=True)
     updated_at = models.DateTimeField(auto_now=True)
@@ -219,6 +228,9 @@ def get_source_identifier(self):
         Get the source identifier string for backward compatibility.
         Returns the source_key or constructs it from components.
         """
+        # Code-defined database query takes precedence
+        if self.source_type == "database" and self.database_query_key:
+            return f"dbquery.{self.database_query_key}"
         if self.source_type == "database" and self.db_schema and self.db_table:
             # Template-based multi-column lookup
             if self.db_template and self.db_columns:
@@ -232,6 +244,10 @@ def get_source_identifier(self):
             return self.source_key
         return self.source_key
 
+    def has_custom_query(self):
+        """Check if this source uses a code-defined database query."""
+        return bool(self.database_query_key)
+
     def has_template(self):
         """Check if this source uses a multi-column template."""
         return bool(self.db_template and self.db_columns)

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "django-forms-workflows"
-version = "0.7.6"
+version = "0.7.7"
 description = "Enterprise-grade, database-driven form builder with approval workflows and external data integration"
 license = "LGPL-3.0-only"
 readme = "README.md"