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]]: 
1240def query_versions_with_pins(
1241    actor_definition_id: str | None = None,
1242    *,
1243    gsm_client: secretmanager.SecretManagerServiceClient | None = None,
1244) -> list[dict[str, Any]]:
1245    """Query connector versions that have at least one pin.
1246
1247    Does NOT join `connector_rollout`, so each version appears exactly once
1248    regardless of how many rollouts reference it.  Includes per-scope pin
1249    breakdown (`actor_pins`, `workspace_pins`, `org_pins`).
1250
1251    Args:
1252        actor_definition_id: Optional connector definition UUID to filter results.
1253            If `None`, returns the global superset across all connectors.
1254        gsm_client: GCP Secret Manager client. If `None`, a new client will be instantiated.
1255
1256    Returns:
1257        List of version dicts ordered by `pin_count` DESC, `created_at` DESC.
1258    """
1259    if actor_definition_id is not None:
1260        return _run_sql_query(
1261            SELECT_VERSIONS_WITH_PINS_BY_DEFINITION,
1262            parameters={"actor_definition_id": actor_definition_id},
1263            query_name="SELECT_VERSIONS_WITH_PINS_BY_DEFINITION",
1264            gsm_client=gsm_client,
1265        )
1266    return _run_sql_query(
1267        SELECT_VERSIONS_WITH_PINS,
1268        parameters=None,
1269        query_name="SELECT_VERSIONS_WITH_PINS",
1270        gsm_client=gsm_client,
1271    )

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.