airbyte_ops_mcp.prod_db_access

Prod DB Access module for querying Airbyte Cloud Prod DB Replica.

This module provides:

  • sql.py: SQL query templates and schema documentation
  • db_engine.py: Database connection and engine management
  • queries.py: Query execution functions
 1# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
 2"""Prod DB Access module for querying Airbyte Cloud Prod DB Replica.
 3
 4This module provides:
 5- sql.py: SQL query templates and schema documentation
 6- db_engine.py: Database connection and engine management
 7- queries.py: Query execution functions
 8"""
 9
10from airbyte_ops_mcp.prod_db_access.db_engine import get_pool
11from airbyte_ops_mcp.prod_db_access.queries import query_versions_with_pins
12from airbyte_ops_mcp.prod_db_access.sql import (
13    SELECT_ACTORS_PINNED_TO_VERSION,
14    SELECT_CONNECTIONS_BY_CONNECTOR,
15    SELECT_CONNECTOR_VERSIONS,
16    SELECT_DATAPLANES_LIST,
17    SELECT_DESTINATION_SUCCESSFUL_SYNCS_FOR_VERSION,
18    SELECT_DESTINATION_SYNC_RESULTS_FOR_VERSION,
19    SELECT_LATEST_RC_COUNTERPART_FAILURES_FOR_DESTINATION_ROLLOUT,
20    SELECT_LATEST_RC_COUNTERPART_FAILURES_FOR_SOURCE_ROLLOUT,
21    SELECT_NEW_CONNECTOR_RELEASES,
22    SELECT_ORG_WORKSPACES,
23    SELECT_SOURCE_SUCCESSFUL_SYNCS_FOR_VERSION,
24    SELECT_SOURCE_SYNC_RESULTS_FOR_VERSION,
25    SELECT_VERSION_ID_BY_TAG,
26    SELECT_VERSION_INFO_BY_ID,
27    SELECT_VERSIONS_WITH_PINS,
28    SELECT_VERSIONS_WITH_PINS_BY_DEFINITION,
29    SELECT_WORKSPACE_INFO,
30)
31
32__all__ = [
33    "SELECT_ACTORS_PINNED_TO_VERSION",
34    "SELECT_CONNECTIONS_BY_CONNECTOR",
35    "SELECT_CONNECTOR_VERSIONS",
36    "SELECT_DATAPLANES_LIST",
37    "SELECT_DESTINATION_SUCCESSFUL_SYNCS_FOR_VERSION",
38    "SELECT_DESTINATION_SYNC_RESULTS_FOR_VERSION",
39    "SELECT_LATEST_RC_COUNTERPART_FAILURES_FOR_DESTINATION_ROLLOUT",
40    "SELECT_LATEST_RC_COUNTERPART_FAILURES_FOR_SOURCE_ROLLOUT",
41    "SELECT_NEW_CONNECTOR_RELEASES",
42    "SELECT_ORG_WORKSPACES",
43    "SELECT_SOURCE_SUCCESSFUL_SYNCS_FOR_VERSION",
44    "SELECT_SOURCE_SYNC_RESULTS_FOR_VERSION",
45    "SELECT_VERSIONS_WITH_PINS",
46    "SELECT_VERSIONS_WITH_PINS_BY_DEFINITION",
47    "SELECT_VERSION_ID_BY_TAG",
48    "SELECT_VERSION_INFO_BY_ID",
49    "SELECT_WORKSPACE_INFO",
50    "get_pool",
51    "query_versions_with_pins",
52]
SELECT_ACTORS_PINNED_TO_VERSION = <sqlalchemy.sql.elements.TextClause object>
SELECT_CONNECTIONS_BY_CONNECTOR = <sqlalchemy.sql.elements.TextClause object>
SELECT_CONNECTOR_VERSIONS = <sqlalchemy.sql.elements.TextClause object>
SELECT_DATAPLANES_LIST = <sqlalchemy.sql.elements.TextClause object>
SELECT_DESTINATION_SUCCESSFUL_SYNCS_FOR_VERSION = <sqlalchemy.sql.elements.TextClause object>
SELECT_DESTINATION_SYNC_RESULTS_FOR_VERSION = <sqlalchemy.sql.elements.TextClause object>
SELECT_LATEST_RC_COUNTERPART_FAILURES_FOR_DESTINATION_ROLLOUT = <sqlalchemy.sql.elements.TextClause object>
SELECT_LATEST_RC_COUNTERPART_FAILURES_FOR_SOURCE_ROLLOUT = <sqlalchemy.sql.elements.TextClause object>
SELECT_NEW_CONNECTOR_RELEASES = <sqlalchemy.sql.elements.TextClause object>
SELECT_ORG_WORKSPACES = <sqlalchemy.sql.elements.TextClause object>
SELECT_SOURCE_SUCCESSFUL_SYNCS_FOR_VERSION = <sqlalchemy.sql.elements.TextClause object>
SELECT_SOURCE_SYNC_RESULTS_FOR_VERSION = <sqlalchemy.sql.elements.TextClause object>
SELECT_VERSIONS_WITH_PINS = <sqlalchemy.sql.elements.TextClause object>
SELECT_VERSIONS_WITH_PINS_BY_DEFINITION = <sqlalchemy.sql.elements.TextClause object>
SELECT_VERSION_ID_BY_TAG = <sqlalchemy.sql.elements.TextClause object>
SELECT_VERSION_INFO_BY_ID = <sqlalchemy.sql.elements.TextClause object>
SELECT_WORKSPACE_INFO = <sqlalchemy.sql.elements.TextClause object>
def get_pool( gsm_client: google.cloud.secretmanager_v1.services.secret_manager_service.client.SecretManagerServiceClient) -> sqlalchemy.engine.base.Engine: 
181def get_pool(
182    gsm_client: secretmanager.SecretManagerServiceClient,
183) -> sqlalchemy.Engine:
184    """Get a SQLAlchemy connection pool for the Airbyte Cloud database.
185
186    The engine is created once and cached for the lifetime of the process so
187    that connection pooling works as intended. Subsequent calls return the
188    same engine instance.
189
190    This function connects with the Cloud SQL Python Connector in public IP mode.
191
192    Args:
193        gsm_client: GCP Secret Manager client for retrieving credentials
194
195    Returns:
196        SQLAlchemy Engine connected to the Prod DB Replica
197    """
198    global _engine
199    if _engine is not None:
200        return _engine
201
202    pg_connection_details = json.loads(
203        _get_secret_value(
204            gsm_client, CONNECTION_RETRIEVER_PG_CONNECTION_DETAILS_SECRET_ID
205        )
206    )
207
208    _engine = sqlalchemy.create_engine(
209        f"postgresql+{PG_DRIVER}://",
210        creator=get_database_creator(pg_connection_details),
211        connect_args={"timeout": DIRECT_CONNECTION_TIMEOUT},
212        pool_size=10,
213        max_overflow=20,
214        pool_timeout=30,
215        pool_recycle=1800,
216    )
217    return _engine

Get a SQLAlchemy connection pool for the Airbyte Cloud database.

The engine is created once and cached for the lifetime of the process so that connection pooling works as intended. Subsequent calls return the same engine instance.

This function connects with the Cloud SQL Python Connector in public IP mode.

Arguments:
  • gsm_client: GCP Secret Manager client for retrieving credentials
Returns:

SQLAlchemy Engine connected to the Prod DB Replica

def query_versions_with_pins( actor_definition_id: str | None = None, *, gsm_client: google.cloud.secretmanager_v1.services.secret_manager_service.client.SecretManagerServiceClient | None = None) -> list[dict[str, typing.Any]]: 
1458def query_versions_with_pins(
1459    actor_definition_id: str | None = None,
1460    *,
1461    gsm_client: secretmanager.SecretManagerServiceClient | None = None,
1462) -> list[dict[str, Any]]:
1463    """Query connector versions that have at least one pin.
1464
1465    Does NOT join `connector_rollout`, so each version appears exactly once
1466    regardless of how many rollouts reference it.  Includes per-scope pin
1467    breakdown (`actor_pins`, `workspace_pins`, `org_pins`).
1468
1469    Args:
1470        actor_definition_id: Optional connector definition UUID to filter results.
1471            If `None`, returns the global superset across all connectors.
1472        gsm_client: GCP Secret Manager client. If `None`, a new client will be instantiated.
1473
1474    Returns:
1475        List of version dicts ordered by `pin_count` DESC, `created_at` DESC.
1476    """
1477    if actor_definition_id is not None:
1478        return _run_sql_query(
1479            SELECT_VERSIONS_WITH_PINS_BY_DEFINITION,
1480            parameters={"actor_definition_id": actor_definition_id},
1481            query_name="SELECT_VERSIONS_WITH_PINS_BY_DEFINITION",
1482            gsm_client=gsm_client,
1483        )
1484    return _run_sql_query(
1485        SELECT_VERSIONS_WITH_PINS,
1486        parameters=None,
1487        query_name="SELECT_VERSIONS_WITH_PINS",
1488        gsm_client=gsm_client,
1489    )

Query connector versions that have at least one pin.

Does NOT join connector_rollout, so each version appears exactly once regardless of how many rollouts reference it. Includes per-scope pin breakdown (actor_pins, workspace_pins, org_pins).

Arguments:
  • actor_definition_id: Optional connector definition UUID to filter results. If None, returns the global superset across all connectors.
  • gsm_client: GCP Secret Manager client. If None, a new client will be instantiated.
Returns:

List of version dicts ordered by pin_count DESC, created_at DESC.