-
Notifications
You must be signed in to change notification settings - Fork 72
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Models for Privacy Center configuration (#4716)
- Loading branch information
Showing
8 changed files
with
258 additions
and
2 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
59 changes: 59 additions & 0 deletions
59
src/fides/api/alembic/migrations/versions/2e9aba76c322_add_privacy_center_config.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,59 @@ | ||
"""add privacy center config | ||
Revision ID: 2e9aba76c322 | ||
Revises: 69e51a460e66 | ||
Create Date: 2024-03-14 22:17:45.544448 | ||
""" | ||
|
||
import sqlalchemy as sa | ||
from alembic import op | ||
from sqlalchemy.dialects import postgresql | ||
|
||
# revision identifiers, used by Alembic. | ||
revision = "2e9aba76c322" | ||
down_revision = "69e51a460e66" | ||
branch_labels = None | ||
depends_on = None | ||
|
||
|
||
def upgrade(): | ||
op.create_table( | ||
"plus_privacy_center_config", | ||
sa.Column("id", sa.String(length=255), nullable=False), | ||
sa.Column( | ||
"created_at", | ||
sa.DateTime(timezone=True), | ||
server_default=sa.text("now()"), | ||
nullable=True, | ||
), | ||
sa.Column( | ||
"updated_at", | ||
sa.DateTime(timezone=True), | ||
server_default=sa.text("now()"), | ||
nullable=True, | ||
), | ||
sa.Column("config", postgresql.JSONB(astext_type=sa.Text()), nullable=False), | ||
sa.Column("single_row", sa.Boolean(), nullable=False), | ||
sa.PrimaryKeyConstraint("id"), | ||
sa.CheckConstraint( | ||
"single_row", name="plus_privacy_center_config_single_row_check" | ||
), | ||
sa.UniqueConstraint( | ||
"single_row", name="plus_privacy_center_config_single_row_unique" | ||
), | ||
) | ||
op.create_index( | ||
op.f("ix_plus_privacy_center_config_id"), | ||
"plus_privacy_center_config", | ||
["id"], | ||
unique=False, | ||
) | ||
|
||
|
||
def downgrade(): | ||
op.drop_index( | ||
op.f("ix_plus_privacy_center_config_id"), | ||
table_name="plus_privacy_center_config", | ||
) | ||
op.drop_table("plus_privacy_center_config") |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,70 @@ | ||
from typing import Any, Dict | ||
|
||
from sqlalchemy import Boolean, CheckConstraint, Column, UniqueConstraint | ||
from sqlalchemy.dialects.postgresql import JSONB | ||
from sqlalchemy.ext.declarative import declared_attr | ||
from sqlalchemy.ext.mutable import MutableDict | ||
from sqlalchemy.orm import Session | ||
|
||
from fides.api.db.base_class import Base, FidesBase | ||
|
||
default_privacy_center_config = { | ||
"actions": [ | ||
{ | ||
"policy_key": "default_access_policy", | ||
"title": "Access your data", | ||
"identity_inputs": {"email": "required"}, | ||
}, | ||
{ | ||
"policy_key": "default_erasure_policy", | ||
"title": "Erase your data", | ||
"identity_inputs": {"email": "required"}, | ||
}, | ||
], | ||
} | ||
|
||
|
||
class PrivacyCenterConfig(Base): | ||
""" | ||
A single-row table used to store the subset of the Privacy Center's config.json that is supported by the Admin UI. | ||
""" | ||
|
||
@declared_attr | ||
def __tablename__(self) -> str: | ||
return "plus_privacy_center_config" | ||
|
||
config = Column( | ||
MutableDict.as_mutable(JSONB), | ||
nullable=False, | ||
) | ||
single_row = Column( | ||
Boolean, | ||
default=True, | ||
nullable=False, | ||
unique=True, | ||
) # used to constrain table to one row | ||
|
||
CheckConstraint("single_row", name="plus_privacy_center_config_single_row_check") | ||
UniqueConstraint("single_row", name="plus_privacy_center_config_single_row_unique") | ||
|
||
@classmethod | ||
def create_or_update( # type: ignore[override] | ||
cls, | ||
db: Session, | ||
*, | ||
data: Dict[str, Any], | ||
) -> FidesBase: | ||
""" | ||
Creates a new config record if one doesn't exist, or updates the existing record. | ||
Here we effectively prevent more than a single record in the table. | ||
""" | ||
existing_record = db.query(cls).first() | ||
if existing_record: | ||
updated_record = existing_record.update( | ||
db=db, | ||
data=data, | ||
) # type: ignore[arg-type] | ||
return updated_record | ||
|
||
return cls.create(db=db, data=data) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,43 @@ | ||
from typing import Any, Dict, List, Literal, Optional | ||
|
||
from pydantic import root_validator | ||
|
||
from fides.api.schemas.base_class import FidesSchema | ||
|
||
RequiredType = Literal["optional", "required"] | ||
|
||
|
||
class IdentityInputs(FidesSchema): | ||
name: Optional[RequiredType] = None | ||
email: Optional[RequiredType] = None | ||
phone: Optional[RequiredType] = None | ||
|
||
|
||
class CustomPrivacyRequestField(FidesSchema): | ||
label: str | ||
required: Optional[bool] = False | ||
default_value: Optional[str] = None | ||
hidden: Optional[bool] = False | ||
|
||
@root_validator | ||
def validate_default_value(cls, values: Dict[str, Any]) -> Dict[str, Any]: | ||
if values.get("hidden") and values.get("default_value") is None: | ||
raise ValueError("default_value is required when hidden is True") | ||
return values | ||
|
||
|
||
class PrivacyRequestOption(FidesSchema): | ||
policy_key: str | ||
title: str | ||
identity_inputs: Optional[IdentityInputs] = None | ||
custom_privacy_request_fields: Optional[Dict[str, CustomPrivacyRequestField]] = None | ||
|
||
class Config: | ||
extra = "ignore" | ||
|
||
|
||
class PrivacyCenterConfig(FidesSchema): | ||
actions: List[PrivacyRequestOption] | ||
|
||
class Config: | ||
extra = "ignore" |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,42 @@ | ||
from sqlalchemy.orm import Session | ||
|
||
from fides.api.models.privacy_center_config import ( | ||
PrivacyCenterConfig, | ||
default_privacy_center_config, | ||
) | ||
|
||
|
||
class TestPrivacyCenterConfig: | ||
def test_create_or_update_keeps_single_record(self, db: Session): | ||
# create the record | ||
config_record = PrivacyCenterConfig.create_or_update( | ||
db=db, data={"config": default_privacy_center_config} | ||
) | ||
assert config_record.config == default_privacy_center_config | ||
assert len(db.query(PrivacyCenterConfig).all()) == 1 | ||
|
||
config_record_db = db.query(PrivacyCenterConfig).first() | ||
assert config_record_db.config == default_privacy_center_config | ||
assert config_record_db.id == config_record.id | ||
|
||
# issue another create_or_update and verify that only the config value changes for the existing row | ||
# no new rows should be created | ||
updated_privacy_center_config = { | ||
"config": { | ||
"actions": [ | ||
{ | ||
"policy_key": "default_access_policy", | ||
"title": "Access your data", | ||
"identity_inputs": {"email": "required"}, | ||
}, | ||
], | ||
} | ||
} | ||
new_config_record = PrivacyCenterConfig.create_or_update( | ||
db=db, data={"config": updated_privacy_center_config} | ||
) | ||
assert len(db.query(PrivacyCenterConfig).all()) == 1 | ||
|
||
new_config_record_db = db.query(PrivacyCenterConfig).first() | ||
assert new_config_record_db.config == updated_privacy_center_config | ||
assert new_config_record_db.id == new_config_record.id |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,28 @@ | ||
import pytest | ||
from pydantic import ValidationError | ||
|
||
from fides.api.schemas.privacy_center_config import ( | ||
CustomPrivacyRequestField, | ||
IdentityInputs, | ||
) | ||
|
||
|
||
class TestPrivacyCenterConfig: | ||
def test_valid_custom_privacy_request_field(self): | ||
CustomPrivacyRequestField(label="Tenant ID", default_value="123", hidden=True) | ||
|
||
def test_custom_privacy_request_field_with_missing_default_value(self): | ||
with pytest.raises(ValidationError) as exc: | ||
CustomPrivacyRequestField(label="Tenant ID", hidden=True) | ||
assert "default_value is required when hidden is True" in str(exc.value) | ||
|
||
def test_custom_privacy_request_fields_with_missing_values(self): | ||
with pytest.raises(ValidationError): | ||
CustomPrivacyRequestField() | ||
|
||
def test_identity_inputs(self): | ||
IdentityInputs(email="required") | ||
|
||
def test_identity_inputs_invalid_value(self): | ||
with pytest.raises(ValidationError): | ||
IdentityInputs(email="[email protected]") |