feat: Add Prefect Cloud pipeline connector#27607
Conversation
|
Hi there 👋 Thanks for your contribution! The OpenMetadata team will review the PR shortly! Once it has been labeled as Let us know if you need any help! |
![gitar-bot[bot]](https://avatars.githubusercontent.com/in/827041?s=60&v=4){kind=small}
|
Hi there 👋 Thanks for your contribution! The OpenMetadata team will review the PR shortly! Once it has been labeled as Let us know if you need any help! |
harshach
added
the
safe to test
|
@itsroshanharry can you make sure to use claude skills to run connector-audit and connector-review. Finally add integration tests with prefect docker and see if we are getting lineage not just the pipeline metadata and also statuses of the pipelines |
|
|
The Python checkstyle failed. Please run You can install the pre-commit hooks with |
Jest test CoverageUI tests summary
|
🔴 Playwright Results — 2 failure(s), 13 flaky✅ 3950 passed · ❌ 2 failed · 🟡 13 flaky · ⏭️ 86 skipped
Genuine Failures (failed on all attempts)❌
|
|
- Change get_pipelines_list() to yield flows instead of returning list - Fixes TypeError from calling len() on generator - Count flows as they're yielded for logging - Prevents memory issues with large number of flows
|
- Change get_pipelines_list() to yield flows instead of returning list - Fixes TypeError from calling len() on generator - Count flows as they're yielded for logging - Fix unit tests to use assertIsNotNone instead of assertTrue(x is not None) - Convert generator to list in unit test before checking length - Prevents memory issues with large number of flows
Code Review ✅ Approved 13 resolved / 13 findingsIntegrates the Prefect Cloud pipeline connector, resolving 13 issues including improper authentication, lack of pagination, and hardcoded configuration values. The implementation is now robust and production-ready. ✅ 13 resolved✅ Bug: test_connection sends requests without Authorization header
✅ Bug: Hardcoded API URL ignores hostPort schema field
✅ Bug: Spoofed User-Agent header is unnecessary and misleading
✅ Performance: Deployments fetched twice per flow (yield_pipeline + lineage)
✅ Edge Case: No pagination: silently drops flows beyond first 200
...and 8 more resolved from earlier reviews OptionsDisplay: compact → Showing less information. Comment with these commands to change:
Was this helpful? React with 👍 / 👎 | Gitar |
|
|
|
|
This PR has had no activity for 30 days and will be closed in 7 days if no further activity occurs. |
IceS2
left a comment
IceS2
left a comment
There was a problem hiding this comment.
Thanks for taking this on.
There's a fair amount to work through before this can land, so I've grouped it. Feel free to reach out if you need anything.
-
Any reason to use httpx instead of requests? We have an idea to migrate to httpx in the future, but for now it would make sense to keep the same pattern as the rest of the code.
-
The connection layer needs to move to BaseConnection. This is the current pattern for all 13 pipeline connectors.
Instead of a module-level get_connection() + test_connection(), define
PrefectConnection(BaseConnection[PrefectConnectionConfig, PrefectClient]) with _get_client() and test_connection(), register it via connection_class in service_spec.py, and drop the test_connection override in metadata.py:498. Moving the HTTP calls into a client.py is part of the same change.
- Four things that will prevent data from landing:
- Either.left expects a StackTraceError, not a bare exception (metadata.py:373, :386). As written, every error path raises a ValidationError. See dagster/metadata.py for the shape.
- yield_pipeline_status needs to yield OMetaPipelineStatus(pipeline_fqn=..., pipeline_status=...) rather than a bare PipelineStatus (metadata.py:283).
- Both filter payloads are off. /deployments/filter wants {"flows": {"id": {"any_": [flow_id]}}}. Docs here. (https://docs.prefect.io/v3/api-ref/rest-api/server/deployments/read-deployments)
- The lineage edges should be table → table with LineageDetails(pipeline=..., source=LineageSource.PipelineLineage), not table → pipeline + pipeline → table (metadata.py:434, :458). airbyte/metadata.py is the canonical example.
- Two security items:
- apiKey should use "format": "password" rather than "password": true (prefectConnection.json:26).Fixing this also lets you delete both hasattr(api_key, "get_secret_value") branches.
- SSL verification is currently disabled by default. get_verify_ssl_fn returns None for no-ssl to mean "use the client default" (i.e. verify), but connection.py:84 maps None → False.
- Worth fixing while you're in there:
- Tags: metadata.py:310 emits TagLabel(source=TagSource.Classification) but nothing overrides yield_tag to create the classification first, so those labels point at nothing. dagster/metadata.py shows the pattern.
- Lineage resolution bypasses _get_table_fqn_from_om (pipeline_service.py), so dbServiceNames from the ingestion config is ignored and users have to hardcode the OM service name into their Prefect tags. Using the helper is what makes the FQN requirement in your PR description go away.
- metadata.py:418 builds the pipeline FQN with string concatenation; fqn.build(...) handles dots and quoting.
- Unknown Prefect states default to Failed (metadata.py:266). Pending is safer.
- _get_deployments is one request per flow and gets called twice per flow (metadata.py:301, :399). Since /deployments/filter takes a list of flow ids, batching a window of ~50 into a bounded LRUCache collapses both problems. It's also unpaginated (hard limit: 50), so a flow with more deployments loses tasks.
- The except Exception blocks in _get_flows / _get_flow_runs /_get_deployments turn a 401 into an empty result, so a bad API key looks like an empty workspace.
Tests. These would all have been caught by an integration test that actually runs. prefecthq/prefect:3-latest should work fine under testcontainers, which the repo already uses (see tests/integration/containers.py). That'd be a much better use of the Docker fixture in conftest.py:43 than the hand-rolled subprocess.run + sleep loop, and it'd let the suite run in CI. On the unit side: we use pytest style (plain classes, bare assert) rather than unittest.TestCase.
Small stuff: missing license headers on init.py and service_spec.py (CI will flag); f-strings in logger.* calls should be lazy %-style; a few imports inside function bodies (metadata.py:247, :429, :453);
Again, thanks for the contribution. The bones are good; it's mostly a matter of routing through the framework's existing helpers rather than around them.
| def get_connection(connection: PrefectConnection) -> httpx.Client: | ||
| """ | ||
| Create an HTTP client for Prefect Cloud or self-hosted Prefect Server. | ||
| """ | ||
| # Validate accountId/workspaceId consistency | ||
| has_account = bool(connection.accountId) | ||
| has_workspace = bool(connection.workspaceId) | ||
| if has_account != has_workspace: | ||
| raise ValueError( | ||
| "Both accountId and workspaceId must be provided for " | ||
| "Prefect Cloud, or both must be empty for self-hosted mode." | ||
| ) | ||
|
|
||
| # Handle both SecretStr and plain string for apiKey | ||
| api_key = connection.apiKey | ||
| if hasattr(api_key, "get_secret_value"): | ||
| api_key_str = api_key.get_secret_value() | ||
| else: | ||
| api_key_str = str(api_key) | ||
|
|
||
| # Build base URL using shared helper | ||
| base_url = _build_base_url(connection) | ||
|
|
||
| headers = { | ||
| "Authorization": f"Bearer {api_key_str}", | ||
| "Content-Type": "application/json", | ||
| "User-Agent": "OpenMetadata/Prefect-Connector", | ||
| } | ||
|
|
||
| # Handle SSL verification (required for enterprise deployments) | ||
| verify_ssl = True # Default to verifying SSL for security | ||
| if connection.verifySSL: | ||
| verify_ssl_fn = get_verify_ssl_fn(connection.verifySSL) | ||
| ssl_result = verify_ssl_fn(connection.sslConfig) | ||
| # None (no-ssl) and False (ignore) both mean disable verification | ||
| verify_ssl = ssl_result if ssl_result is not None else False | ||
|
|
||
| return httpx.Client( | ||
| base_url=base_url, headers=headers, timeout=30, verify=verify_ssl | ||
| ) |
There was a problem hiding this comment.
Can you please use the BaseConnection? We are moving away from these module level functions and using the BaseConnection to handle it
| def test_connection( | ||
| metadata: OpenMetadata, | ||
| client: httpx.Client, | ||
| service_connection: PrefectConnection, | ||
| base_url: str = None, | ||
| headers: dict = None, | ||
| automation_workflow: AutomationWorkflow = None, | ||
| ) -> None: | ||
| """ | ||
| Test connection to Prefect Cloud by fetching flows. | ||
| """ | ||
|
|
||
| def custom_test_connection(client: httpx.Client) -> None: | ||
| # Test using POST /flows/filter as per Prefect 3.x API | ||
| # Use provided base_url and headers if available (from metadata.py) | ||
| # Otherwise construct them (when called from get_connection) | ||
| if base_url and headers: | ||
| url = f"{base_url}/flows/filter" | ||
| response = client.post(url, headers=headers, json={"limit": 1, "offset": 0}) | ||
| else: | ||
| # Fallback: client already has base_url and headers configured | ||
| response = client.post("/flows/filter", json={"limit": 1, "offset": 0}) | ||
| response.raise_for_status() | ||
|
|
||
| test_fn = {"GetFlows": custom_test_connection} | ||
| test_connection_steps( | ||
| metadata=metadata, | ||
| service_type="Prefect", | ||
| test_fn=test_fn, | ||
| automation_workflow=automation_workflow, | ||
| ) |
There was a problem hiding this comment.
While changing to BaseConnection, could you also migrate to the new TestConnection that provides diagnostics on different errors? You can check an example here: #29767
| from metadata.ingestion.source.pipeline.prefect.metadata import PrefectSource | ||
| from metadata.utils.service_spec import BaseSpec | ||
|
|
||
| ServiceSpec = BaseSpec(metadata_source_class=PrefectSource) |
There was a problem hiding this comment.
Once BaseConnection is set, it needs to be added here
| # Validate accountId/workspaceId consistency | ||
| has_account = bool(self.service_connection.accountId) | ||
| has_workspace = bool(self.service_connection.workspaceId) | ||
| if has_account != has_workspace: | ||
| raise InvalidSourceException( | ||
| "Both accountId and workspaceId must be provided for " | ||
| "Prefect Cloud, or both must be empty for self-hosted mode." | ||
| ) | ||
|
|
||
| # Use shared base URL builder | ||
| self.base_url = _build_base_url(self.service_connection) | ||
|
|
||
| # Handle both SecretStr and plain string for apiKey | ||
| api_key = self.service_connection.apiKey | ||
| if hasattr(api_key, "get_secret_value"): | ||
| api_key_str = api_key.get_secret_value() | ||
| else: | ||
| api_key_str = str(api_key) | ||
|
|
||
| self.headers = { | ||
| "Authorization": f"Bearer {api_key_str}", | ||
| "Content-Type": "application/json", | ||
| "User-Agent": "OpenMetadata/Prefect-Connector", | ||
| } |
There was a problem hiding this comment.
I feel these should be handled in the connection.py when creating the connection and not here, no?
There was a problem hiding this comment.
PipelineServiceSource should already build the client, you should not need to rebuild the url and headers and pass it manually unless I am missing something
| def _get_flows(self) -> Iterable[dict]: | ||
| """ | ||
| Fetch flows from Prefect using pagination with generator pattern. | ||
| Yields flows one at a time to avoid loading all flows into memory. | ||
| """ | ||
| offset = 0 | ||
| limit = 200 # Prefect API maximum per request | ||
|
|
||
| try: | ||
| while True: | ||
| response = self.connection.post( | ||
| f"{self.base_url}/flows/filter", | ||
| headers=self.headers, | ||
| json={"limit": limit, "offset": offset}, | ||
| ) | ||
| response.raise_for_status() | ||
| flows = response.json() | ||
|
|
||
| if not flows: | ||
| break | ||
|
|
||
| # Yield each flow individually instead of accumulating | ||
| yield from flows | ||
|
|
||
| # If we got fewer than limit, we've reached the end | ||
| if len(flows) < limit: | ||
| break | ||
|
|
||
| offset += limit | ||
|
|
There was a problem hiding this comment.
I would extract the API calls to a client.py module like it is done in other connectors.
| for source_fqn in sources: | ||
| from metadata.generated.schema.entity.data.table import Table | ||
|
|
||
| source_table = self.metadata.get_by_name(entity=Table, fqn=source_fqn) |
There was a problem hiding this comment.
source_fqn here is the raw tag suffix — whatever the user typed after om-source:. The base class provides _get_table_fqn_from_om (pipeline_service.py:320) for exactly this resolution:
def _get_table_fqn_from_om(self, table_details: TableDetails) -> Optional[str]:
result = None
services = self.get_db_service_names()
for db_service in services:
result = fqn.build(
metadata=self.metadata, entity_type=Table,
service_name=db_service,
database_name=table_details.database,
schema_name=table_details.schema,
table_name=table_details.name,
)
if result:
return result
raise FQNNotFoundException(f"Table FQN not found for table: {table_details} within services: {services}")
| for dest_fqn in destinations: | ||
| from metadata.generated.schema.entity.data.table import Table | ||
|
|
||
| dest_table = self.metadata.get_by_name(entity=Table, fqn=dest_fqn) |
There was a problem hiding this comment.
Same as the note on using the utility method I have done above
| service_fqn = self.context.get().pipeline_service | ||
| pipeline_fqn = f"{service_fqn}.{flow_name}" |
There was a problem hiding this comment.
Should use the fqn.build utility
|
|
||
| # Create lineage edges for source tables | ||
| for source_fqn in sources: | ||
| from metadata.generated.schema.entity.data.table import Table |
|
|
||
| # Create lineage edges for destination tables | ||
| for dest_fqn in destinations: | ||
| from metadata.generated.schema.entity.data.table import Table |
4 participants
Describe your changes:
Fixes #26656
This PR adds a new pipeline connector for Prefect Cloud, enabling OpenMetadata to ingest pipeline metadata, run history, and lineage from Prefect 3.x workspaces.
What changes did I make?
Why did I make them?
How did I test my changes?
Type of change:
Checklist:
feat: Add Prefect Cloud pipeline connectorNew feature checklist:
Summary
This PR adds a new pipeline connector for Prefect Cloud, enabling OpenMetadata to ingest pipeline metadata, run history, and lineage from Prefect 3.x workspaces.
Features
om-source:,om-destination:)Files Changed
Backend/Schema Files:
openmetadata-spec/src/main/resources/json/schema/entity/services/connections/pipeline/prefectConnection.json- Connection configuration schemaopenmetadata-spec/src/main/resources/json/schema/entity/services/pipelineService.json- Added Prefect to service type enumPython Connector:
ingestion/src/metadata/ingestion/source/pipeline/prefect/__init__.py- Package initingestion/src/metadata/ingestion/source/pipeline/prefect/metadata.py- Main connector logicingestion/src/metadata/ingestion/source/pipeline/prefect/connection.py- Connection testingingestion/src/metadata/ingestion/source/pipeline/prefect/service_spec.py- Service specificationUI Files:
openmetadata-ui/src/main/resources/ui/src/utils/PipelineServiceUtils.ts- Added Prefect to UI utilitiesTests:
ingestion/tests/unit/topology/pipeline/test_prefect.py- Comprehensive unit tests (6 tests, all passing)Configuration Example
Lineage Detection
The connector supports tag-based lineage detection with two formats:
Recommended Format (Prefixed)
Tag your Prefect flows or deployments with the
om-prefix to avoid conflicts with other tagging conventions:om-source:service.database.schema.table- for source tables (full FQN)om-destination:service.database.schema.table- for destination tables (full FQN)Example:
Legacy Format (Backward Compatible)
The connector also supports the legacy format without prefix:
source:service.database.schema.tabledestination:service.database.schema.tableExample:
Features:
Best Practices:
om-source:,om-destination:) to avoid conflictsservice.database.schema.tableordatabase.schema.tableImportant Notes:
mysql.warehouse.sales.ordersorwarehouse.sales.orders). Simple table names likeorderswill not be found.Testing
All unit tests pass:
Documentation Requirements
The following documentation needs to be added to the
openmetadata-docsrepository:1. Connector Documentation
Create folder:
openmetadata-docs/content/v1.12-SNAPSHOT/connectors/pipeline/prefect/Files needed:
index.mdx- UI configuration guideyaml.mdx- YAML configuration guide2. Update Connector Lists
Add Prefect to:
openmetadata-docs/content/v1.12-SNAPSHOT/menu.mdxopenmetadata-docs/content/v1.12-SNAPSHOT/connectors/index.mdxopenmetadata-docs/content/v1.12-SNAPSHOT/connectors/pipeline/index.mdxopenmetadata-docs/partials/v1.12.x/connectors-list.mdx3. Add Connector Logo
Upload Prefect logo to:
openmetadata-docs/public/images/connectors/prefect.png4. Add Installation Images
Create directory:
openmetadata-docs/public/images/v1.12/connectors/prefect/Add screenshots for installation steps
Implementation Details
API Compatibility
POST /flows/filterinstead ofGET /flows)Data Mapping
Status Mapping
Breaking Changes
None - this is a new connector.
Checklist
🚀 Future Enhancements
This PR delivers a stable, production-ready V1 connector with comprehensive functionality. Potential enhancements for future versions include:
Artifact-Based Lineage
While the current version uses a robust tag-based approach (optimal for most Prefect users and consistent with the Prefect 3.x API), future iterations could explore automatic lineage tracking via Prefect's artifacts API for even deeper integration. This would allow lineage to be captured programmatically within flow code rather than through tags.
Self-Hosted Prefect Server Support
The initial focus is on Prefect Cloud (the most common deployment). Testing and validation for local/self-hosted Prefect Server instances will follow in a future update. The connector architecture is designed to support both with minimal changes (primarily URL configuration).
Enhanced Metadata
Future versions could ingest additional Prefect metadata such as:
Pagination for Large Workspaces
The current implementation fetches up to 200 flows per request (Prefect API limit). For workspaces with >200 flows, pagination logic could be added to fetch all flows across multiple requests.
These enhancements are intentionally scoped out of V1 to ensure a stable, well-tested foundation that can be extended incrementally based on community feedback.
Related Issues
Closes #26656 - New Pipeline Connectors (Prefect implementation)
This PR implements the Prefect connector as requested in the "New Pipeline Connectors" issue. Prefect was listed as one of the modern workflow orchestration tools to be added to OpenMetadata's connector ecosystem.
Additional Notes
Summary by Gitar
_get_flowsto stream pipeline results instead of loading all into memory.This will update automatically on new commits.