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_NEW_CONNECTOR_RELEASES,
20    SELECT_ORG_WORKSPACES,
21    SELECT_SOURCE_SUCCESSFUL_SYNCS_FOR_VERSION,
22    SELECT_SOURCE_SYNC_RESULTS_FOR_VERSION,
23    SELECT_VERSION_ID_BY_TAG,
24    SELECT_VERSION_INFO_BY_ID,
25    SELECT_VERSIONS_WITH_PINS,
26    SELECT_VERSIONS_WITH_PINS_BY_DEFINITION,
27    SELECT_WORKSPACE_INFO,
28)
29
30__all__ = [
31    "SELECT_ACTORS_PINNED_TO_VERSION",
32    "SELECT_CONNECTIONS_BY_CONNECTOR",
33    "SELECT_CONNECTOR_VERSIONS",
34    "SELECT_DATAPLANES_LIST",
35    "SELECT_DESTINATION_SUCCESSFUL_SYNCS_FOR_VERSION",
36    "SELECT_DESTINATION_SYNC_RESULTS_FOR_VERSION",
37    "SELECT_NEW_CONNECTOR_RELEASES",
38    "SELECT_ORG_WORKSPACES",
39    "SELECT_SOURCE_SUCCESSFUL_SYNCS_FOR_VERSION",
40    "SELECT_SOURCE_SYNC_RESULTS_FOR_VERSION",
41    "SELECT_VERSIONS_WITH_PINS",
42    "SELECT_VERSIONS_WITH_PINS_BY_DEFINITION",
43    "SELECT_VERSION_ID_BY_TAG",
44    "SELECT_VERSION_INFO_BY_ID",
45    "SELECT_WORKSPACE_INFO",
46    "get_pool",
47    "query_versions_with_pins",
48]
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_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]]: 
1422def query_versions_with_pins(
1423    actor_definition_id: str | None = None,
1424    *,
1425    gsm_client: secretmanager.SecretManagerServiceClient | None = None,
1426) -> list[dict[str, Any]]:
1427    """Query connector versions that have at least one pin.
1428
1429    Does NOT join `connector_rollout`, so each version appears exactly once
1430    regardless of how many rollouts reference it.  Includes per-scope pin
1431    breakdown (`actor_pins`, `workspace_pins`, `org_pins`).
1432
1433    Args:
1434        actor_definition_id: Optional connector definition UUID to filter results.
1435            If `None`, returns the global superset across all connectors.
1436        gsm_client: GCP Secret Manager client. If `None`, a new client will be instantiated.
1437
1438    Returns:
1439        List of version dicts ordered by `pin_count` DESC, `created_at` DESC.
1440    """
1441    if actor_definition_id is not None:
1442        return _run_sql_query(
1443            SELECT_VERSIONS_WITH_PINS_BY_DEFINITION,
1444            parameters={"actor_definition_id": actor_definition_id},
1445            query_name="SELECT_VERSIONS_WITH_PINS_BY_DEFINITION",
1446            gsm_client=gsm_client,
1447        )
1448    return _run_sql_query(
1449        SELECT_VERSIONS_WITH_PINS,
1450        parameters=None,
1451        query_name="SELECT_VERSIONS_WITH_PINS",
1452        gsm_client=gsm_client,
1453    )

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.