ScyllaDB Connector

CHANGELOG.md

@@ -18,6 +18,7 @@ The types of changes are:
 ## [Unreleased](https://github.com/ethyca/fides/compare/2.38.0...main)
 
 ### Added
+- Adds the start of the Scylla DB Integration [#4946](https://github.com/ethyca/fides/pull/4946)
 - Added model and data migrations and CRUD-layer operations for property-specific messaging [#4901](https://github.com/ethyca/fides/pull/4901)
 - Added option in FidesJS SDK to only disable notice-served API [#4965](https://github.com/ethyca/fides/pull/4965)
 - External ID support for consent management [#4927](https://github.com/ethyca/fides/pull/4927)

clients/admin-ui/src/features/datastore-connections/constants.ts

@@ -46,6 +46,7 @@ export const CONNECTION_TYPE_LOGO_MAP = new Map<ConnectionType, string>([
   [ConnectionType.MYSQL, "mysql.svg"],
   [ConnectionType.POSTGRES, "postgres.svg"],
   [ConnectionType.REDSHIFT, "redshift.svg"],
+  [ConnectionType.SCYLLA, "scylla.svg"],
   [ConnectionType.SNOWFLAKE, "snowflake.svg"],
   [ConnectionType.SOVRN, "sovrn.svg"],
   [ConnectionType.TIMESCALE, "timescaledb.svg"],

clients/admin-ui/src/types/api/models/ConnectionType.ts

@@ -23,6 +23,7 @@ export enum ConnectionType {
   MANUAL_WEBHOOK = "manual_webhook",
   TIMESCALE = "timescale",
   FIDES = "fides",
+  SCYLLA = "scylla",
   GENERIC_ERASURE_EMAIL = "generic_erasure_email",
   GENERIC_CONSENT_EMAIL = "generic_consent_email",
 }

docker/docker-compose.integration-scylladb.yml

@@ -0,0 +1,15 @@
+services:
+  scylladb_example:
+    image: bitnami/scylladb:6.0.0
+    container_name: scylladb_example
+    environment:
+      - SCYLLADB_PASSWORD_SEEDER=yes
+      - SCYLLADB_USER=scylla_user
+      - SCYLLADB_PASSWORD=scylla_pass
+      - SCYLLADB_AUTHENTICATOR=PasswordAuthenticator
+    expose:
+      - 9042
+    volumes:
+      - ./docker/sample_data/scylla:/docker-entrypoint-initdb.d
+    ports:
+      - "9042:9042"

docker/sample_data/scylla/scylla_example.cql

@@ -0,0 +1,35 @@
+CREATE KEYSPACE IF NOT EXISTS app_keyspace WITH REPLICATION = {'class': 'SimpleStrategy', 'replication_factor': 3};
+USE app_keyspace;
+
+CREATE TABLE if not exists users (
+    user_id INT PRIMARY KEY,
+    name TEXT,
+    age INT,
+    email TEXT
+);
+
+INSERT INTO users (user_id, name, age, email) VALUES (1, 'John', 41, 'customer-1@example.com') IF NOT EXISTS;
+INSERT INTO users (user_id, name, age, email) VALUES (2, 'Jane', 38, 'jane@example.com') IF NOT EXISTS;
+INSERT INTO users (user_id, name, age, email) VALUES (3, 'Marguerite', 27, 'customer-2@example.com') IF NOT EXISTS;
+INSERT INTO users (user_id, name, age, email) VALUES (4, 'Lafayette', 55, 'customer-3@example.com') IF NOT EXISTS;
+INSERT INTO users (user_id, name, age, email) VALUES (5, 'Manuel', 23, 'customer-4@example.com') IF NOT EXISTS;
+
+
+CREATE KEYSPACE IF NOT EXISTS vendors_keyspace WITH REPLICATION = {'class': 'SimpleStrategy', 'replication_factor': 3};
+USE vendors_keyspace;
+
+CREATE TABLE if not exists vendors (
+    vendor_id INT PRIMARY KEY,
+    vendor_name TEXT,
+    vendor_address TEXT,
+    primary_contact_name TEXT,
+    primary_contact_email TEXT,
+    supplier_type TEXT
+);
+
+INSERT INTO vendors (vendor_id, vendor_name, vendor_address, primary_contact_name, primary_contact_email, supplier_type) VALUES (1, 'A+ Tile Supplies', '1810 Test Town, TX', 'Elliot Trace', 'employee-1@example.com', 'building supplies') IF NOT EXISTS;
+INSERT INTO vendors (vendor_id, vendor_name, vendor_address, primary_contact_name, primary_contact_email, supplier_type) VALUES (2, 'Carpeting and More', '3421 Test City, TX', 'Chris Osbourne', 'employee-2@example.com', 'building supplies') IF NOT EXISTS;
+INSERT INTO vendors (vendor_id, vendor_name, vendor_address, primary_contact_name, primary_contact_email, supplier_type) VALUES (3, 'Sunshine Bakery', '2394 Gottlieb Station, Romeoton, MS', 'Fatima Sultani', 'employee-3@example.com', 'coffee and tea') IF NOT EXISTS;
+INSERT INTO vendors (vendor_id, vendor_name, vendor_address, primary_contact_name, primary_contact_email, supplier_type) VALUES (4, 'Sweetbrew', '609 Shanahan Points, Guillermotown, NC', 'Xavier Gutierrez', 'employee-4@example.com', 'coffee and tea') IF NOT EXISTS;
+INSERT INTO vendors (vendor_id, vendor_name, vendor_address, primary_contact_name, primary_contact_email, supplier_type) VALUES (5, 'Artevista', '595 Blick Drive, Demetricebury, MO', 'Lana Anderson', 'employee-5@example.com', 'art supplies') IF NOT EXISTS;
+INSERT INTO vendors (vendor_id, vendor_name, vendor_address, primary_contact_name, primary_contact_email, supplier_type) VALUES (6, 'CanvasCrafter', '1410 Parkway Street, San Diego, CA', 'Madelyn Houston', 'employee-6@example.com', 'art supplies') IF NOT EXISTS;

noxfiles/run_infrastructure.py

@@ -19,6 +19,7 @@
     "mongodb",
     "mariadb",
     "timescale",
+    "scylladb",
 ]
 EXTERNAL_DATASTORE_CONFIG = {
     "snowflake": [
@@ -84,6 +85,8 @@ def run_infrastructure(
         if datastore in DOCKERFILE_DATASTORES
     ]
 
+    _run_cmd_or_err(f'echo "Docker datastores {docker_datastores}"')
+
     # Configure docker compose path
     path: str = get_path_for_datastores(datastores, remote_debug)
 

pyproject.toml

@@ -38,6 +38,7 @@ module = [
   "boto3.*",
   "botocore.*",
   "bson.*",
+  "cassandra.*",
   "celery.*",
   "citext.*",
   "dask.*",
@@ -150,7 +151,7 @@ good-names="_,i,setUp,tearDown,maxDiff,default_app_config"
 ignore="migrations,tests"
 
 [tool.pylint.whitelist]
-extension-pkg-whitelist = ["pydantic", "zlib"]
+extension-pkg-whitelist = ["pydantic", "zlib", "cassandra"]
 
 ############
 ## Pytest ##
@@ -191,6 +192,7 @@ markers = [
     "integration_dynamodb",
     "integration_saas",
     "integration_saas_override",
+    "integration_scylladb",
     "unit_saas"
 ]
 asyncio_mode = "auto"

requirements.txt

@@ -48,6 +48,7 @@ pyyaml==6.0.1
 redis==3.5.3
 rich-click==1.6.1
 sendgrid==6.9.7
+scylla-driver==3.26.8
 slowapi==0.1.8
 snowflake-sqlalchemy==1.5.1
 sqlalchemy[asyncio]==1.4.27

src/fides/api/alembic/migrations/versions/3304082a6cee_scylla_connection_type.py

@@ -0,0 +1,46 @@
+"""scylla_connection_type
+
+Revision ID: 3304082a6cee
+Revises: 5fe01e730171
+Create Date: 2024-06-03 19:54:20.907724
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "3304082a6cee"
+down_revision = "5fe01e730171"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # Add to ConnectionType enum
+    op.execute("alter type connectiontype rename to connectiontype_old")
+    op.execute(
+        "create type connectiontype as enum('mongodb', 'mysql', 'https', 'snowflake', 'redshift', 'mssql', 'mariadb', 'bigquery', 'saas', 'manual', 'manual_webhook', 'timescale', 'fides', 'sovrn', 'attentive', 'dynamodb', 'postgres', 'generic_consent_email', 'generic_erasure_email', 'scylla')"
+    )
+    op.execute(
+        (
+            "alter table connectionconfig alter column connection_type type connectiontype using "
+            "connection_type::text::connectiontype"
+        )
+    )
+    op.execute("drop type connectiontype_old")
+
+
+def downgrade():
+    # Remove 'scylla' from ConnectionType
+    op.execute("alter type connectiontype rename to connectiontype_old")
+    op.execute(
+        "create type connectiontype as enum('mongodb', 'mysql', 'https', 'snowflake', 'redshift', 'mssql', 'mariadb', 'bigquery', 'saas', 'manual', 'manual_webhook', 'timescale', 'fides', 'sovrn', 'attentive', 'dynamodb', 'postgres', 'generic_consent_email', 'generic_erasure_email')"
+    )
+    op.execute(
+        (
+            "alter table connectionconfig alter column connection_type type connectiontype using "
+            "connection_type::text::connectiontype"
+        )
+    )
+    op.execute("drop type connectiontype_old")

src/fides/api/api/v1/endpoints/generate.py

@@ -47,6 +47,7 @@ class ValidTargets(str, Enum):
     OKTA = "okta"
     BIGQUERY = "bigquery"
     DYNAMODB = "dynamodb"
+    SCYLLADB = "scylla"
 
 
 class GenerateTypes(str, Enum):

src/fides/api/models/connectionconfig.py

@@ -53,6 +53,7 @@ class ConnectionType(enum.Enum):
     fides = "fides"
     generic_erasure_email = "generic_erasure_email"  # Run after the traversal
     generic_consent_email = "generic_consent_email"  # Run after the traversal
+    scylla = "scylla"
 
     @property
     def human_readable(self) -> str:
@@ -76,6 +77,7 @@ def human_readable(self) -> str:
             ConnectionType.postgres.value: "PostgreSQL",
             ConnectionType.redshift.value: "Amazon Redshift",
             ConnectionType.saas.value: "SaaS",
+            ConnectionType.scylla.value: "Scylla DB",
             ConnectionType.snowflake.value: "Snowflake",
             ConnectionType.sovrn.value: "Sovrn",
             ConnectionType.timescale.value: "TimescaleDB",

src/fides/api/schemas/connection_configuration/__init__.py

@@ -86,6 +86,9 @@
 from fides.api.schemas.connection_configuration.connection_secrets_saas import (
     SaaSSchemaFactory as SaaSSchemaFactory,
 )
+from fides.api.schemas.connection_configuration.connection_secrets_scylla import (
+    ScyllaSchema, ScyllaDocsSchema,
+)
 from fides.api.schemas.connection_configuration.connection_secrets_snowflake import (
     SnowflakeDocsSchema as SnowflakeDocsSchema,
 )
@@ -123,6 +126,7 @@
     ConnectionType.postgres.value: PostgreSQLSchema,
     ConnectionType.redshift.value: RedshiftSchema,
     ConnectionType.saas.value: SaaSSchema,
+    ConnectionType.scylla.value: ScyllaSchema,
     ConnectionType.snowflake.value: SnowflakeSchema,
     ConnectionType.sovrn.value: SovrnSchema,
     ConnectionType.timescale.value: TimescaleSchema,
@@ -171,4 +175,5 @@ def get_connection_secrets_schema(
     FidesDocsSchema,
     SovrnDocsSchema,
     DynamoDBDocsSchema,
+    ScyllaDocsSchema
 ]

src/fides/api/schemas/connection_configuration/connection_secrets_scylla.py

@@ -0,0 +1,46 @@
+from typing import List, Optional
+
+from pydantic import Field
+
+from fides.api.schemas.base_class import NoValidationSchema
+from fides.api.schemas.connection_configuration.connection_secrets import (
+    ConnectionConfigSecretsSchema,
+)
+
+
+class ScyllaSchema(ConnectionConfigSecretsSchema):
+    """Schema to validate the secrets needed to connect to a Scylla Database"""
+
+    host: str = Field(
+        title="Host",
+        description="The hostname or IP address of the server where the database is running.",
+    )
+    port: int = Field(
+        9042,
+        title="Port",
+        description="The network port number on which the server is listening for incoming connections (default: 9042).",
+    )
+    username: str = Field(
+        title="Username",
+        description="The user account used to authenticate and access the database.",
+    )
+    password: str = Field(
+        title="Password",
+        description="The password used to authenticate and access the database.",
+        sensitive=True,
+    )
+    keyspace: Optional[str] = Field(
+        title="Keyspace",
+        description="The keyspace used.",
+        sensitive=True,
+    )
+
+    _required_components: List[str] = [
+        "host",
+        "username",
+        "password",
+    ]
+
+
+class ScyllaDocsSchema(ScyllaSchema, NoValidationSchema):
+    """Scylla Secrets Schema for API docs"""

src/fides/api/service/connectors/__init__.py

@@ -31,6 +31,7 @@
     MongoDBConnector as MongoDBConnector,
 )
 from fides.api.service.connectors.saas_connector import SaaSConnector as SaaSConnector
+from fides.api.service.connectors.scylla_connector import ScyllaConnector
 from fides.api.service.connectors.sql_connector import (
     BigQueryConnector as BigQueryConnector,
 )
@@ -70,6 +71,7 @@
     ConnectionType.postgres.value: PostgreSQLConnector,
     ConnectionType.redshift.value: RedshiftConnector,
     ConnectionType.saas.value: SaaSConnector,
+    ConnectionType.scylla.value: ScyllaConnector,
     ConnectionType.snowflake.value: SnowflakeConnector,
     ConnectionType.sovrn.value: SovrnConnector,
     ConnectionType.timescale.value: TimescaleConnector,

src/fides/api/service/connectors/scylla_connector.py

@@ -0,0 +1,97 @@
+from typing import Any, Dict, List, Optional
+
+import cassandra
+from cassandra.auth import PlainTextAuthProvider
+from cassandra.cluster import Cluster
+from loguru import logger
+
+from fides.api.common_exceptions import ConnectionException
+from fides.api.graph.execution import ExecutionNode
+from fides.api.models.connectionconfig import ConnectionTestStatus
+from fides.api.models.policy import Policy
+from fides.api.models.privacy_request import PrivacyRequest, RequestTask
+from fides.api.schemas.connection_configuration.connection_secrets_scylla import (
+    ScyllaSchema,
+)
+from fides.api.service.connectors.base_connector import BaseConnector
+from fides.api.service.connectors.query_config import QueryConfig
+from fides.api.util.collection_util import Row
+
+
+class ScyllaConnector(BaseConnector):
+    """Scylla Connector"""
+
+    def build_uri(self) -> str:
+        """
+        Builds URI
+        """
+
+    def create_client(self) -> Cluster:
+        """Returns a Scylla cluster"""
+
+        config = ScyllaSchema(**self.configuration.secrets or {})
+
+        auth_provider = PlainTextAuthProvider(
+            username=config.username, password=config.password
+        )
+        cluster = Cluster([config.host], port=config.port, auth_provider=auth_provider)
+        return cluster
+
+    def query_config(self, node: ExecutionNode) -> QueryConfig[Any]:
+        pass
+
+    def test_connection(self) -> Optional[ConnectionTestStatus]:
+        """
+        Connects to the scylla database and issues a trivial query
+        """
+        logger.info("Starting test connection to {}", self.configuration.key)
+        config = ScyllaSchema(**self.configuration.secrets or {})
+        cluster = self.client()
+
+        try:
+            with cluster.connect(
+                config.keyspace if config.keyspace else None
+            ) as client:
+                client.execute("select now() from system.local")
+        except cassandra.cluster.NoHostAvailable as exc:
+            if "Unable to connect to any servers using keyspace" in str(exc):
+                raise ConnectionException("Unknown keyspace.")
+            try:
+                error = list(exc.errors.values())[0]
+            except Exception:
+                raise ConnectionException("No host available.")
+
+            if isinstance(error, cassandra.AuthenticationFailed):
+                raise ConnectionException("Authentication failed.")
+
+            raise ConnectionException("No host available.")
+
+        except cassandra.protocol.SyntaxException:
+            raise ConnectionException("Syntax exception.")
+        except Exception:
+            raise ConnectionException("Connection Error connecting to Scylla DB.")
+
+        return ConnectionTestStatus.succeeded
+
+    def retrieve_data(
+        self,
+        node: ExecutionNode,
+        policy: Policy,
+        privacy_request: PrivacyRequest,
+        request_task: RequestTask,
+        input_data: Dict[str, List[Any]],
+    ) -> List[Row]:
+        """Retrieve scylla data - not yet implemented"""
+
+    def mask_data(
+        self,
+        node: ExecutionNode,
+        policy: Policy,
+        privacy_request: PrivacyRequest,
+        request_task: RequestTask,
+        rows: List[Row],
+    ) -> int:
+        """Execute a masking request - not yet implemented"""
+
+    def close(self) -> None:
+        """Close any held resources"""

tests/fixtures/application_fixtures.py

@@ -182,6 +182,11 @@
             integration_config, "dynamodb_example.aws_secret_access_key"
         ),
     },
+    "scylla_example": {
+        "host": pydash.get(integration_config, "scylladb_example.server"),
+        "username": pydash.get(integration_config, "scylladb_example.username"),
+        "password": pydash.get(integration_config, "scylladb_example.password"),
+    },
 }