Enterprise Connector and REST Catalog Boundary Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Move database enterprise connectors into Data Factory, add a real PostgreSQL connector, and integrate REST Catalog into the Metadata page without changing existing connector APIs or security boundaries.

Architecture: PostgreSQL becomes a versioned read-only catalog connector that reuses the current data-source manager and pool. The combined frontend is split into one allowlist-driven operations component, a database-only Data Factory wrapper, and a REST-only Metadata tab. The existing enterprise-connector route path and backend APIs remain stable.

Tech Stack: Python 3, Flask, SQLAlchemy, pytest, Vue 2, Vuetify, Vuex, JavaScript, Docker Compose.

File Structure

Create app/core/connectors/builtin/postgresql.py and its byte-identical deployment mirror for the PostgreSQL connector.
Modify both builtin/init.py files to register PostgreSQL.
Create tests/test_postgresql_enterprise_connector.py for backend behavior.
Modify frontend/src/router/routes.js for navigation ownership.
Create tests/test_enterprise_connector_metadata_navigation_contract.py for exact route ownership and order.
Create frontend/src/components/connectors/ConnectorOperations.vue for allowlist-driven shared operations.
Simplify frontend/src/views/dataGovernance/development/enterpriseConnectors.vue into the database wrapper.
Move the current metadata implementation into components/metadataManagement.vue.
Create components/externalCatalogAccess.vue and a new metadata/index.vue tab container.
Create tests/test_enterprise_connector_metadata_frontend_contract.py for filtering, tabs, and permission behavior.

Task 1: Add a Real PostgreSQL Enterprise Connector

Files:

Create: app/core/connectors/builtin/postgresql.py
Create: deployment/app/core/connectors/builtin/postgresql.py
Modify: app/core/connectors/builtin/init.py
Modify: deployment/app/core/connectors/builtin/init.py
Test: tests/test_postgresql_enterprise_connector.py
[ ] Step 1: Write the failing backend test

Create a fake SQLAlchemy-compatible connection provider and verify registration, projection, scope, checkpoints, cancellation, resume, and secret-free evidence.

from contextlib import contextmanager

import pytest

from app.core.connectors.builtin import register_builtin_connectors
from app.core.connectors.builtin.postgresql import PostgreSQLConnector
from app.core.connectors.errors import ConnectorCancelledError
from app.core.connectors.registry import ConnectorRegistry
from app.core.connectors.sdk import OperationRequest


ROWS = [
    {
        "schema_name": "public",
        "asset_name": "orders",
        "asset_type": "BASE TABLE",
        "column_name": "id",
        "ordinal_position": 1,
        "data_type": "uuid",
        "is_nullable": "NO",
        "column_default": None,
        "column_comment": "order identifier",
    },
    {
        "schema_name": "internal",
        "asset_name": "jobs",
        "asset_type": "BASE TABLE",
        "column_name": "id",
        "ordinal_position": 1,
        "data_type": "bigint",
        "is_nullable": "NO",
        "column_default": None,
        "column_comment": None,
    },
]


class Result:
    def mappings(self):
        return self

    def all(self):
        return ROWS


class Connection:
    def execute(self, statement, parameters):
        assert "information_schema.columns" in str(statement)
        assert parameters == {}
        return Result()


@contextmanager
def connection_provider(source_uid, purpose, **kwargs):
    assert source_uid == "source-postgres"
    assert purpose == "metadata_collection"
    assert kwargs == {}
    yield Connection()


def request(operation="discover", **changes):
    values = {
        "source_uid": "source-postgres",
        "operation": operation,
        "config": {"credential_ref": "env:DATAOPS_CONNECTOR_POSTGRES"},
        "scope": {"include_schemas": ["public"]},
        "checkpoint": {},
    }
    values.update(changes)
    return OperationRequest(**values)


def test_postgresql_connector_registers_and_projects_catalog():
    registry = register_builtin_connectors(
        ConnectorRegistry(), connection_provider=connection_provider
    )
    connector = registry.resolve("postgresql", "1.0.0")
    assert isinstance(connector, PostgreSQLConnector)
    assert connector.manifest.display_name == "PostgreSQL"
    assert connector.manifest.capabilities == (
        "discover", "snapshot", "incremental", "cancel", "resume", "evidence"
    )
    result = connector.discover(request())
    assert [item["asset_key"] for item in result.records] == [
        "source-postgres:public.orders"
    ]
    assert result.checkpoint["snapshot_summary"]["record_count"] == 1
    assert result.evidence["query_kind"] == "read_only_metadata"
    assert "DATAOPS_CONNECTOR_POSTGRES" not in str(result)


def test_postgresql_connector_incremental_cancel_resume_and_evidence():
    connector = PostgreSQLConnector(connection_provider)
    first = connector.snapshot(request("snapshot"))
    incremental = connector.incremental(
        request("incremental", checkpoint=first.checkpoint)
    )
    assert incremental.evidence["diff"]["changed"] is False
    assert connector.resume(request("resume", checkpoint=first.checkpoint)).status == "succeeded"
    assert connector.cancel(request("cancel")).status == "cancelled"
    assert connector.evidence(request("evidence")).evidence["secret_material"] is False
    with pytest.raises(ConnectorCancelledError):
        connector.discover(request(cancel_probe=lambda: True))

Step 2: Run RED

Run:

PYTHONPATH=. .venv/bin/pytest -q tests/test_postgresql_enterprise_connector.py

Expected: collection fails because app.core.connectors.builtin.postgresql does not exist.

Step 3: Implement PostgreSQL

Create PostgreSQLConnector as a ReadOnlyCatalogConnector. Use information_schema.columns joined to information_schema.tables and pg_catalog description tables; exclude pg_catalog and information_schema. Use connector_id postgresql, version 1.0.0, display name PostgreSQL, DATABASE_CONFIG_SCHEMA, and capabilities discover/snapshot/incremental/cancel/resume/evidence. health() and compatibility() check importlib.util.find_spec("psycopg2").

connection_provider = dependencies.get("connection_provider")
registry.register(OracleConnector(connection_provider))
registry.register(PostgreSQLConnector(connection_provider))
registry.register(SqlServerConnector(connection_provider))

Add PostgreSQLConnector to all. Copy both changed backend files byte-for-byte into deployment/app.

Step 4: Run GREEN and mirrors

PYTHONPATH=. .venv/bin/pytest -q   tests/test_postgresql_enterprise_connector.py   tests/test_phase3_wp03_enterprise_connectors.py
cmp app/core/connectors/builtin/postgresql.py deployment/app/core/connectors/builtin/postgresql.py
cmp app/core/connectors/builtin/__init__.py deployment/app/core/connectors/builtin/__init__.py

Expected: all tests pass and both comparisons exit 0.

Step 5: Commit

git add app/core/connectors/builtin/postgresql.py   deployment/app/core/connectors/builtin/postgresql.py   app/core/connectors/builtin/__init__.py   deployment/app/core/connectors/builtin/__init__.py   tests/test_postgresql_enterprise_connector.py
git commit -m "feat: add postgresql enterprise connector"

Task 2: Move Enterprise Connectors into Data Factory

Files:

Modify: frontend/src/router/routes.js
Create: tests/test_enterprise_connector_metadata_navigation_contract.py
[ ] Step 1: Write the failing navigation test

Use the scoped route-block pattern from tests/test_enterprise_identity_navigation_contract.py. Assert:

def test_enterprise_connectors_belong_only_to_data_factory():
    routes = ROUTES.read_text(encoding="utf-8")
    research = route_block(routes, "data-governance", 4)
    factory = route_block(routes, "dataFactory", 4)
    assert routes.count("name: 'enterpriseConnectors'") == 1
    assert "name: 'enterpriseConnectors'" not in research
    assert "name: 'enterpriseConnectors'" in factory
    block = route_block(factory, "enterpriseConnectors", 8)
    assert "path: '/data-governance/development/enterprise-connectors'" in block
    assert "component: 'dataGovernance/development/enterpriseConnectors'" in block
    assert "permissions: ['connectors:read']" in block


def test_data_factory_order_keeps_legacy_entries_independent():
    routes = ROUTES.read_text(encoding="utf-8")
    factory = route_block(routes, "dataFactory", 4)
    names = [
        "enterpriseConnectors",
        "productionLineDeployment",
        "productionLine",
        "workflowIndex",
        "dataObservability",
        "connectionPoolIndex",
    ]
    positions = [factory.index(f"name: '{name}'") for name in names]
    assert positions == sorted(positions)

Step 2: Run RED

PYTHONPATH=. .venv/bin/pytest -q tests/test_enterprise_connector_metadata_navigation_contract.py

Expected: enterpriseConnectors is still under Data Research.

Step 3: Move the route

Move the existing object unchanged into dataFactory.children immediately before productionLineDeployment. Add sort: 4. Preserve path, name, component, icon, label, and connectors:read. Do not change n8n paths, targets, names, or components.

Step 4: Run GREEN

PYTHONPATH=. .venv/bin/pytest -q   tests/test_enterprise_connector_metadata_navigation_contract.py   tests/test_data_factory_orchestration_navigation_contract.py   tests/test_enterprise_identity_menu_filter_contract.py

Expected: all pass.

Step 5: Commit

git add frontend/src/router/routes.js   tests/test_enterprise_connector_metadata_navigation_contract.py
git commit -m "feat: move enterprise connectors to data factory"

Task 3: Extract the Database Connector Operations Surface

Files:

Create: frontend/src/components/connectors/ConnectorOperations.vue
Modify: frontend/src/views/dataGovernance/development/enterpriseConnectors.vue
Create: tests/test_enterprise_connector_metadata_frontend_contract.py
[ ] Step 1: Write failing contracts

def test_shared_operations_is_allowlist_driven_and_fail_closed():
    source = SHARED.read_text(encoding="utf-8")
    assert "connectorIds" in source
    assert "normalizedConnectorIds" in source
    assert "filter(item => allowed.has(item.connector_id))" in source
    assert "if (!this.normalizedConnectorIds.length)" in source


def test_enterprise_surface_is_database_only():
    source = DATABASE.read_text(encoding="utf-8")
    assert "['oracle', 'postgresql', 'sqlserver']" in source
    assert "当前开放范围仅限数据库访问" in source
    assert "文件目录、对象存储、API 与消息系统" in source
    assert "rest-catalog" not in source

Step 2: Run RED

PYTHONPATH=. .venv/bin/pytest -q tests/test_enterprise_connector_metadata_frontend_contract.py

Expected: the shared component is missing and the old page still mixes REST Catalog.

Step 3: Implement ConnectorOperations.vue

Move the current manifest cards, run table, compatibility, Dry-run, cancel, and resume behavior into the shared component. Add strict props:

props: {
  connectorIds: {
    type: Array,
    required: true,
    validator: value => value.length > 0 && value.every(
      item => typeof item === 'string' && /^[a-z][a-z0-9_-]{2,63}$/.test(item)
    )
  },
  mode: {
    type: String,
    required: true,
    validator: value => ['database', 'rest-catalog'].includes(value)
  }
}

Normalize the allowlist and fail closed:

normalizedConnectorIds () {
  if (!Array.isArray(this.connectorIds)) return []
  return [...new Set(this.connectorIds.filter(
    item => typeof item === 'string' && /^[a-z][a-z0-9_-]{2,63}$/.test(item)
  ))]
}

In loadAll(), return without API calls when the list is empty. Otherwise load manifests and runs, then filter both with the same Set. Clear stale arrays in the catch block. Render base_url and allowed_host only in rest-catalog mode. Preserve existing connectors:operate/manage checks and human Dry-run rules.

Step 4: Make enterpriseConnectors.vue a database wrapper

Render the heading, the enterprise-UAT warning, and:

<v-alert type="info" outlined>
  当前开放范围仅限数据库访问。文件目录、对象存储、API 与消息系统等来源将在后续版本扩展开发。
</v-alert>
<connector-operations
  :connector-ids="['oracle', 'postgresql', 'sqlserver']"
  mode="database"
/>

Step 5: Run GREEN and build

PYTHONPATH=. .venv/bin/pytest -q   tests/test_enterprise_connector_metadata_frontend_contract.py   tests/test_phase3_wp03_enterprise_connectors.py
npm --prefix frontend run build

Expected: tests pass and build exits 0.

Step 6: Commit

git add frontend/src/components/connectors/ConnectorOperations.vue   frontend/src/views/dataGovernance/development/enterpriseConnectors.vue   tests/test_enterprise_connector_metadata_frontend_contract.py
git commit -m "refactor: isolate database connector operations"

Task 4: Integrate REST Catalog into Metadata Governance

Files:

Move: frontend/src/views/dataGovernance/metadata/index.vue
Create: frontend/src/views/dataGovernance/metadata/components/metadataManagement.vue
Create: frontend/src/views/dataGovernance/metadata/components/externalCatalogAccess.vue
Create: frontend/src/views/dataGovernance/metadata/index.vue
Modify: tests/test_enterprise_connector_metadata_frontend_contract.py
[ ] Step 1: Add failing tab and permission tests

def test_metadata_has_management_and_external_catalog_tabs():
    source = METADATA.read_text(encoding="utf-8")
    assert source.count("<v-tab") == 2
    assert "元数据管理" in source
    assert "外部目录接入" in source
    assert "permissions.includes('connectors:read')" in source
    assert 'v-if="canReadConnectors"' in source


def test_external_catalog_is_rest_only_and_management_is_preserved():
    external = EXTERNAL.read_text(encoding="utf-8")
    management = MANAGEMENT.read_text(encoding="utf-8")
    assert "['rest-catalog']" in external
    assert 'mode="rest-catalog"' in external
    assert "Oracle" not in external and "SQL Server" not in external
    for marker in ("新增元数据", "handleSubmit", "handleDelete", "getMetaDataList"):
        assert marker in management

Step 2: Run RED

PYTHONPATH=. .venv/bin/pytest -q tests/test_enterprise_connector_metadata_frontend_contract.py

Expected: the tab wrapper and REST component are missing.

Step 3: Preserve metadata management

Move the complete current index.vue into components/metadataManagement.vue. Only fix relative imports caused by the move:

import FilterList from '../../components/Filter'
import Edit from './edit'
import AddAudit from './addAudit'

Keep its existing CRUD, audit, status, dialogs, assistant comment, and Data Research Center link unchanged.

Step 4: Create externalCatalogAccess.vue

<template>
  <div class="pt-4">
    <v-alert type="info" outlined>
      REST Catalog 用于从受控 HTTPS 元数据目录同步资产，不采集任意业务 API 数据。
    </v-alert>
    <v-alert type="warning" outlined>
      真实企业目录地址、允许域名、凭据和 UAT 仍需外部提供。
    </v-alert>
    <connector-operations :connector-ids="['rest-catalog']" mode="rest-catalog" />
  </div>
</template>

Import ConnectorOperations and register it as the only child component.

Step 5: Recreate metadata/index.vue as a permission-aware wrapper

<template>
  <div class="pa-3 white">
    <v-tabs v-model="tab" show-arrows>
      <v-tab>元数据管理</v-tab>
      <v-tab v-if="canReadConnectors">外部目录接入</v-tab>
    </v-tabs>
    <v-tabs-items v-model="tab">
      <v-tab-item><metadata-management /></v-tab-item>
      <v-tab-item v-if="canReadConnectors"><external-catalog-access /></v-tab-item>
    </v-tabs-items>
  </div>
</template>

Compute permissions from this.$store.state.user.userInfo.permissions and return permissions.includes('connectors:read') from canReadConnectors. The v-if must prevent mounting and API loading for unauthorized users.

Step 6: Run GREEN and build

PYTHONPATH=. .venv/bin/pytest -q   tests/test_enterprise_connector_metadata_frontend_contract.py   tests/data_research/test_data_research_frontend_contract.py   tests/test_phase3_wp03_enterprise_connectors.py   tests/test_frontend_rbac_contract.py
npm --prefix frontend run build

Expected: all pass and build exits 0.

Step 7: Commit

git add frontend/src/views/dataGovernance/metadata/index.vue   frontend/src/views/dataGovernance/metadata/components/metadataManagement.vue   frontend/src/views/dataGovernance/metadata/components/externalCatalogAccess.vue   tests/test_enterprise_connector_metadata_frontend_contract.py
git commit -m "feat: integrate rest catalog with metadata governance"

Task 5: Run Affected Regression and Static Gates

Files: Verification only unless a failure is proven to be caused by Tasks 1–4.

Step 1: Run the complete affected suite

PYTHONPATH=. .venv/bin/pytest -q   tests/test_postgresql_enterprise_connector.py   tests/test_enterprise_connector_metadata_navigation_contract.py   tests/test_enterprise_connector_metadata_frontend_contract.py   tests/test_phase3_wp03_enterprise_connectors.py   tests/data_research/test_data_research_frontend_contract.py   tests/test_data_factory_orchestration_navigation_contract.py   tests/test_enterprise_identity_menu_filter_contract.py   tests/test_frontend_rbac_contract.py   tests/test_architecture_artifacts.py

Expected: all collected tests pass and this change introduces no skips.

Step 2: Run existing real PostgreSQL integration

PYTHONPATH=. .venv/bin/pytest -q tests/integration/test_phase3_wp03_connectors_postgres.py

Expected: pass using the repository's documented Docker test database. If the fixture reports the database environment is unavailable, record the exact blocker and do not report this command as passed.

Step 3: Run build, compile, mirror, and diff gates

npm --prefix frontend run build
.venv/bin/python -m py_compile   app/core/connectors/builtin/postgresql.py   deployment/app/core/connectors/builtin/postgresql.py   tests/test_postgresql_enterprise_connector.py   tests/test_enterprise_connector_metadata_navigation_contract.py   tests/test_enterprise_connector_metadata_frontend_contract.py
cmp app/core/connectors/builtin/postgresql.py deployment/app/core/connectors/builtin/postgresql.py
cmp app/core/connectors/builtin/__init__.py deployment/app/core/connectors/builtin/__init__.py
git diff --check
git status --short

Expected: all commands exit 0. Preserve existing untracked architecture assets and do not stage them.

Step 4: Commit only proven gate corrections

If an affected contract required a real update, stage its exact files and commit:

git commit -m "test: align connector metadata boundary contracts"

Do not create an empty commit.

Task 6: Deploy and Perform Local UAT

Files: No source changes expected.

Step 1: Rebuild local backend and frontend

docker compose -f deploy/docker/docker-compose.yml up -d --build backend frontend

Expected: build exits 0 and frontend/backend become healthy.

Step 2: Verify services and deep links

docker compose -f deploy/docker/docker-compose.yml ps
curl -fsS http://localhost:15500/api/system/health
curl -fsS http://localhost:18183/data-governance/development/enterprise-connectors >/dev/null
curl -fsS http://localhost:18183/data-governance/metadata >/dev/null

Expected: health reports status=healthy and both SPA links return 200.

Step 3: Perform authenticated browser UAT

Verify all nine conditions:

Data Research no longer lists 企业连接器.
Data Factory lists 企业连接器 before 数据生产线投产.
Enterprise Connectors shows exactly Oracle, PostgreSQL, and SQL Server.
The database-only scope notice is visible.
REST Catalog is absent from that page.
Metadata shows 元数据管理 and 外部目录接入 for connectors:read.
External Catalog Access shows REST Catalog and no database connectors.
Both direct links survive refresh.
A metadata user without connectors:read retains metadata management and does not mount external-catalog content.

Do not execute a connector against a real external system during this navigation UAT. Compatibility or Dry-run requires a separately approved test endpoint and credential.

Step 4: Run final independent review

PASS requires zero Critical and zero Important findings. Fix each valid finding with a new RED/GREEN cycle and repeat affected review.

Step 5: Record final branch state

git log --oneline -8
git status --short
git diff --check

Expected: feature files committed, unrelated untracked user assets untouched, and no push or production deployment without separate instruction.

2026-08-19-enterprise-connector-metadata-boundary.md 21 KB History Raw

Enterprise Connector and REST Catalog Boundary Implementation Plan

File Structure

Task 1: Add a Real PostgreSQL Enterprise Connector

Task 2: Move Enterprise Connectors into Data Factory

Task 3: Extract the Database Connector Operations Surface

Task 4: Integrate REST Catalog into Metadata Governance

Task 5: Run Affected Regression and Static Gates

Task 6: Deploy and Perform Local UAT

2026-08-19-enterprise-connector-metadata-boundary.md 21 KB

History Raw