Add data_purposes support to datasets

.fides/db_dataset.yml

@@ -697,6 +697,9 @@ dataset:
     - name: data_categories
       data_categories:
       - system.operations
+    - name: data_purposes
+      data_categories:
+      - system.operations
     - name: description
       data_categories:
       - system.operations

changelog/7674-dataset-data-purposes.yaml

@@ -0,0 +1,4 @@
+type: Changed
+description: Updated fideslang to 3.1.3 to add `data_purposes` support at dataset, collection, field, and sub-field levels
+pr: 7674
+labels: []

pyproject.toml

@@ -55,7 +55,7 @@ dependencies = [
     "fastapi-cli~=0.0.16",
     "fastapi-pagination[sqlalchemy]==0.15.0",
     "fastapi[all]==0.123.3",
-    "fideslang==3.1.4a1",
+    "fideslang==3.1.3",
     "firebase-admin==5.3.0",
     "flower==2.0.1",
     "google-auth==2.48.0",

src/fides/api/alembic/migrations/versions/xx_2026_03_25_1500_c7e3a9b1d4f2_add_data_purposes_to_datasets.py

@@ -0,0 +1,27 @@
+"""add data_purposes column to ctl_datasets
+
+Revision ID: c7e3a9b1d4f2
+Revises: 25ffa8bf6a95
+Create Date: 2026-03-25 15:00:00.000000
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "c7e3a9b1d4f2"
+down_revision = "25ffa8bf6a95"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "ctl_datasets",
+        sa.Column("data_purposes", sa.ARRAY(sa.String()), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("ctl_datasets", "data_purposes")

src/fides/api/models/sql_models.py

@@ -406,6 +406,7 @@ class Dataset(Base, FidesBase):
 
     meta = Column(JSON)
     data_categories = Column(ARRAY(String))
+    data_purposes = Column(ARRAY(String))
     collections = Column(JSON)
     fides_meta = Column(JSON)
 

src/fides/service/connection/merge_configs_util.py

@@ -328,14 +328,17 @@ def merge_datasets(
     """
     stored_dataset_copy = copy.deepcopy(stored_dataset)
     upcoming_dataset_copy = copy.deepcopy(upcoming_dataset)
+    customer_dataset_copy = copy.deepcopy(customer_dataset)
 
     normalized_stored_dataset = normalize_dataset(stored_dataset_copy, "stored")
     normalized_upcoming_dataset = normalize_dataset(upcoming_dataset_copy, "upcoming")
+    normalized_customer_dataset = normalize_dataset(customer_dataset_copy, "customer")
     if normalized_stored_dataset is None or normalized_upcoming_dataset is None:
         return upcoming_dataset
 
     stored_dataset_copy = normalized_stored_dataset
     upcoming_dataset_copy = normalized_upcoming_dataset
+    customer_dataset = normalized_customer_dataset or customer_dataset
 
     # convert dataset to yaml string and then replace the instance key placeholder
     wrapped_dataset = {"dataset": [stored_dataset_copy]}