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 (
12    query_versions_with_pins,
13)
14from airbyte_ops_mcp.prod_db_access.sql import (
15    SELECT_ACTORS_PINNED_TO_VERSION,
16    SELECT_CONNECTIONS_BY_CONNECTOR,
17    SELECT_CONNECTOR_VERSIONS,
18    SELECT_DATAPLANES_LIST,
19    SELECT_DESTINATION_SUCCESSFUL_SYNCS_FOR_VERSION,
20    SELECT_DESTINATION_SYNC_RESULTS_FOR_VERSION,
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_NEW_CONNECTOR_RELEASES",
40    "SELECT_ORG_WORKSPACES",
41    "SELECT_SOURCE_SUCCESSFUL_SYNCS_FOR_VERSION",
42    "SELECT_SOURCE_SYNC_RESULTS_FOR_VERSION",
43    "SELECT_VERSIONS_WITH_PINS",
44    "SELECT_VERSIONS_WITH_PINS_BY_DEFINITION",
45    "SELECT_VERSION_ID_BY_TAG",
46    "SELECT_VERSION_INFO_BY_ID",
47    "SELECT_WORKSPACE_INFO",
48    "get_pool",
49    "query_versions_with_pins",
50]
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]]: 
1186def query_versions_with_pins(
1187    actor_definition_id: str | None = None,
1188    *,
1189    gsm_client: secretmanager.SecretManagerServiceClient | None = None,
1190) -> list[dict[str, Any]]:
1191    """Query connector versions that have at least one pin.
1192
1193    Does NOT join `connector_rollout`, so each version appears exactly once
1194    regardless of how many rollouts reference it.  Includes per-scope pin
1195    breakdown (`actor_pins`, `workspace_pins`, `org_pins`).
1196
1197    Args:
1198        actor_definition_id: Optional connector definition UUID to filter results.
1199            If `None`, returns the global superset across all connectors.
1200        gsm_client: GCP Secret Manager client. If `None`, a new client will be instantiated.
1201
1202    Returns:
1203        List of version dicts ordered by `pin_count` DESC, `created_at` DESC.
1204    """
1205    if actor_definition_id is not None:
1206        return _run_sql_query(
1207            SELECT_VERSIONS_WITH_PINS_BY_DEFINITION,
1208            parameters={"actor_definition_id": actor_definition_id},
1209            query_name="SELECT_VERSIONS_WITH_PINS_BY_DEFINITION",
1210            gsm_client=gsm_client,
1211        )
1212    return _run_sql_query(
1213        SELECT_VERSIONS_WITH_PINS,
1214        parameters=None,
1215        query_name="SELECT_VERSIONS_WITH_PINS",
1216        gsm_client=gsm_client,
1217    )

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.