airbyte.cloud

 56@dataclass
 57class CloudWorkspace:
 58    """A remote workspace on the Airbyte Cloud.
 59
 60    By overriding `api_root`, you can use this class to interact with self-managed Airbyte
 61    instances, both OSS and Enterprise.
 62    """
 63
 64    workspace_id: str
 65    client_id: SecretString
 66    client_secret: SecretString
 67    api_root: str = api_util.CLOUD_API_ROOT
 68
 69    def __post_init__(self) -> None:
 70        """Ensure that the client ID and secret are handled securely."""
 71        self.client_id = SecretString(self.client_id)
 72        self.client_secret = SecretString(self.client_secret)
 73
 74    @property
 75    def workspace_url(self) -> str | None:
 76        """The URL of the workspace."""
 77        return f"{self.api_root}/workspaces/{self.workspace_id}"
 78
 79    # Test connection and creds
 80
 81    def connect(self) -> None:
 82        """Check that the workspace is reachable and raise an exception otherwise.
 83
 84        Note: It is not necessary to call this method before calling other operations. It
 85              serves primarily as a simple check to ensure that the workspace is reachable
 86              and credentials are correct.
 87        """
 88        _ = api_util.get_workspace(
 89            api_root=self.api_root,
 90            workspace_id=self.workspace_id,
 91            client_id=self.client_id,
 92            client_secret=self.client_secret,
 93        )
 94        print(f"Successfully connected to workspace: {self.workspace_url}")
 95
 96    # Get sources, destinations, and connections
 97
 98    def get_connection(
 99        self,
100        connection_id: str,
101    ) -> CloudConnection:
102        """Get a connection by ID.
103
104        This method does not fetch data from the API. It returns a `CloudConnection` object,
105        which will be loaded lazily as needed.
106        """
107        return CloudConnection(
108            workspace=self,
109            connection_id=connection_id,
110        )
111
112    def get_source(
113        self,
114        source_id: str,
115    ) -> CloudSource:
116        """Get a source by ID.
117
118        This method does not fetch data from the API. It returns a `CloudSource` object,
119        which will be loaded lazily as needed.
120        """
121        return CloudSource(
122            workspace=self,
123            connector_id=source_id,
124        )
125
126    def get_destination(
127        self,
128        destination_id: str,
129    ) -> CloudDestination:
130        """Get a destination by ID.
131
132        This method does not fetch data from the API. It returns a `CloudDestination` object,
133        which will be loaded lazily as needed.
134        """
135        return CloudDestination(
136            workspace=self,
137            connector_id=destination_id,
138        )
139
140    # Deploy sources and destinations
141
142    def deploy_source(
143        self,
144        name: str,
145        source: Source,
146        *,
147        unique: bool = True,
148        random_name_suffix: bool = False,
149    ) -> CloudSource:
150        """Deploy a source to the workspace.
151
152        Returns the newly deployed source.
153
154        Args:
155            name: The name to use when deploying.
156            source: The source object to deploy.
157            unique: Whether to require a unique name. If `True`, duplicate names
158                are not allowed. Defaults to `True`.
159            random_name_suffix: Whether to append a random suffix to the name.
160        """
161        source_config_dict = source.get_config().copy()
162        source_config_dict["sourceType"] = source.name.replace("source-", "")
163
164        if random_name_suffix:
165            name += f" (ID: {text_util.generate_random_suffix()})"
166
167        if unique:
168            existing = self.list_sources(name=name)
169            if existing:
170                raise exc.AirbyteDuplicateResourcesError(
171                    resource_type="source",
172                    resource_name=name,
173                )
174
175        deployed_source = api_util.create_source(
176            name=name,
177            api_root=self.api_root,
178            workspace_id=self.workspace_id,
179            config=source_config_dict,
180            client_id=self.client_id,
181            client_secret=self.client_secret,
182        )
183        return CloudSource(
184            workspace=self,
185            connector_id=deployed_source.source_id,
186        )
187
188    def deploy_destination(
189        self,
190        name: str,
191        destination: Destination | dict[str, Any],
192        *,
193        unique: bool = True,
194        random_name_suffix: bool = False,
195    ) -> CloudDestination:
196        """Deploy a destination to the workspace.
197
198        Returns the newly deployed destination ID.
199
200        Args:
201            name: The name to use when deploying.
202            destination: The destination to deploy. Can be a local Airbyte `Destination` object or a
203                dictionary of configuration values.
204            unique: Whether to require a unique name. If `True`, duplicate names
205                are not allowed. Defaults to `True`.
206            random_name_suffix: Whether to append a random suffix to the name.
207        """
208        if isinstance(destination, Destination):
209            destination_conf_dict = destination.get_config().copy()
210            destination_conf_dict["destinationType"] = destination.name.replace("destination-", "")
211            # raise ValueError(destination_conf_dict)
212        else:
213            destination_conf_dict = destination.copy()
214            if "destinationType" not in destination_conf_dict:
215                raise exc.PyAirbyteInputError(
216                    message="Missing `destinationType` in configuration dictionary.",
217                )
218
219        if random_name_suffix:
220            name += f" (ID: {text_util.generate_random_suffix()})"
221
222        if unique:
223            existing = self.list_destinations(name=name)
224            if existing:
225                raise exc.AirbyteDuplicateResourcesError(
226                    resource_type="destination",
227                    resource_name=name,
228                )
229
230        deployed_destination = api_util.create_destination(
231            name=name,
232            api_root=self.api_root,
233            workspace_id=self.workspace_id,
234            config=destination_conf_dict,  # Wants a dataclass but accepts dict
235            client_id=self.client_id,
236            client_secret=self.client_secret,
237        )
238        return CloudDestination(
239            workspace=self,
240            connector_id=deployed_destination.destination_id,
241        )
242
243    def permanently_delete_source(
244        self,
245        source: str | CloudSource,
246    ) -> None:
247        """Delete a source from the workspace.
248
249        You can pass either the source ID `str` or a deployed `Source` object.
250        """
251        if not isinstance(source, (str, CloudSource)):
252            raise exc.PyAirbyteInputError(
253                message="Invalid source type.",
254                input_value=type(source).__name__,
255            )
256
257        api_util.delete_source(
258            source_id=source.connector_id if isinstance(source, CloudSource) else source,
259            api_root=self.api_root,
260            client_id=self.client_id,
261            client_secret=self.client_secret,
262        )
263
264    # Deploy and delete destinations
265
266    def permanently_delete_destination(
267        self,
268        destination: str | CloudDestination,
269    ) -> None:
270        """Delete a deployed destination from the workspace.
271
272        You can pass either the `Cache` class or the deployed destination ID as a `str`.
273        """
274        if not isinstance(destination, (str, CloudDestination)):
275            raise exc.PyAirbyteInputError(
276                message="Invalid destination type.",
277                input_value=type(destination).__name__,
278            )
279
280        api_util.delete_destination(
281            destination_id=(
282                destination if isinstance(destination, str) else destination.destination_id
283            ),
284            api_root=self.api_root,
285            client_id=self.client_id,
286            client_secret=self.client_secret,
287        )
288
289    # Deploy and delete connections
290
291    def deploy_connection(
292        self,
293        connection_name: str,
294        *,
295        source: CloudSource | str,
296        selected_streams: list[str],
297        destination: CloudDestination | str,
298        table_prefix: str | None = None,
299    ) -> CloudConnection:
300        """Create a new connection between an already deployed source and destination.
301
302        Returns the newly deployed connection object.
303
304        Args:
305            connection_name: The name of the connection.
306            source: The deployed source. You can pass a source ID or a CloudSource object.
307            destination: The deployed destination. You can pass a destination ID or a
308                CloudDestination object.
309            table_prefix: Optional. The table prefix to use when syncing to the destination.
310            selected_streams: The selected stream names to sync within the connection.
311        """
312        if not selected_streams:
313            raise exc.PyAirbyteInputError(
314                guidance="You must provide `selected_streams` when creating a connection."
315            )
316
317        source_id: str = source if isinstance(source, str) else source.connector_id
318        destination_id: str = (
319            destination if isinstance(destination, str) else destination.connector_id
320        )
321
322        deployed_connection = api_util.create_connection(
323            name=connection_name,
324            source_id=source_id,
325            destination_id=destination_id,
326            api_root=self.api_root,
327            workspace_id=self.workspace_id,
328            selected_stream_names=selected_streams,
329            prefix=table_prefix or "",
330            client_id=self.client_id,
331            client_secret=self.client_secret,
332        )
333
334        return CloudConnection(
335            workspace=self,
336            connection_id=deployed_connection.connection_id,
337            source=deployed_connection.source_id,
338            destination=deployed_connection.destination_id,
339        )
340
341    def permanently_delete_connection(
342        self,
343        connection: str | CloudConnection,
344        *,
345        cascade_delete_source: bool = False,
346        cascade_delete_destination: bool = False,
347    ) -> None:
348        """Delete a deployed connection from the workspace."""
349        if connection is None:
350            raise ValueError("No connection ID provided.")
351
352        if isinstance(connection, str):
353            connection = CloudConnection(
354                workspace=self,
355                connection_id=connection,
356            )
357
358        api_util.delete_connection(
359            connection_id=connection.connection_id,
360            api_root=self.api_root,
361            workspace_id=self.workspace_id,
362            client_id=self.client_id,
363            client_secret=self.client_secret,
364        )
365
366        if cascade_delete_source:
367            self.permanently_delete_source(source=connection.source_id)
368        if cascade_delete_destination:
369            self.permanently_delete_destination(destination=connection.destination_id)
370
371    # List sources, destinations, and connections
372
373    def list_connections(
374        self,
375        name: str | None = None,
376        *,
377        name_filter: Callable | None = None,
378    ) -> list[CloudConnection]:
379        """List connections by name in the workspace."""
380        connections = api_util.list_connections(
381            api_root=self.api_root,
382            workspace_id=self.workspace_id,
383            name=name,
384            name_filter=name_filter,
385            client_id=self.client_id,
386            client_secret=self.client_secret,
387        )
388        return [
389            CloudConnection(
390                workspace=self,
391                connection_id=connection.connection_id,
392                source=None,
393                destination=None,
394            )
395            for connection in connections
396            if name is None or connection.name == name
397        ]
398
399    def list_sources(
400        self,
401        name: str | None = None,
402        *,
403        name_filter: Callable | None = None,
404    ) -> list[CloudSource]:
405        """List all sources in the workspace."""
406        sources = api_util.list_sources(
407            api_root=self.api_root,
408            workspace_id=self.workspace_id,
409            name=name,
410            name_filter=name_filter,
411            client_id=self.client_id,
412            client_secret=self.client_secret,
413        )
414        return [
415            CloudSource(
416                workspace=self,
417                connector_id=source.source_id,
418            )
419            for source in sources
420            if name is None or source.name == name
421        ]
422
423    def list_destinations(
424        self,
425        name: str | None = None,
426        *,
427        name_filter: Callable | None = None,
428    ) -> list[CloudDestination]:
429        """List all destinations in the workspace."""
430        destinations = api_util.list_destinations(
431            api_root=self.api_root,
432            workspace_id=self.workspace_id,
433            name=name,
434            name_filter=name_filter,
435            client_id=self.client_id,
436            client_secret=self.client_secret,
437        )
438        return [
439            CloudDestination(
440                workspace=self,
441                connector_id=destination.destination_id,
442            )
443            for destination in destinations
444            if name is None or destination.name == name
445        ]

 20class CloudConnection:
 21    """A connection is an extract-load (EL) pairing of a source and destination in Airbyte Cloud.
 22
 23    You can use a connection object to run sync jobs, retrieve logs, and manage the connection.
 24    """
 25
 26    def __init__(
 27        self,
 28        workspace: CloudWorkspace,
 29        connection_id: str,
 30        source: str | None = None,
 31        destination: str | None = None,
 32    ) -> None:
 33        """It is not recommended to create a `CloudConnection` object directly.
 34
 35        Instead, use `CloudWorkspace.get_connection()` to create a connection object.
 36        """
 37        self.connection_id = connection_id
 38        """The ID of the connection."""
 39
 40        self.workspace = workspace
 41        """The workspace that the connection belongs to."""
 42
 43        self._source_id = source
 44        """The ID of the source."""
 45
 46        self._destination_id = destination
 47        """The ID of the destination."""
 48
 49        self._connection_info: ConnectionResponse | None = None
 50        """The connection info object. (Cached.)"""
 51
 52        self._cloud_source_object: CloudSource | None = None
 53        """The source object. (Cached.)"""
 54
 55        self._cloud_destination_object: CloudDestination | None = None
 56        """The destination object. (Cached.)"""
 57
 58    def _fetch_connection_info(self) -> ConnectionResponse:
 59        """Populate the connection with data from the API."""
 60        return api_util.get_connection(
 61            workspace_id=self.workspace.workspace_id,
 62            connection_id=self.connection_id,
 63            api_root=self.workspace.api_root,
 64            client_id=self.workspace.client_id,
 65            client_secret=self.workspace.client_secret,
 66        )
 67
 68    # Properties
 69
 70    @property
 71    def source_id(self) -> str:
 72        """The ID of the source."""
 73        if not self._source_id:
 74            if not self._connection_info:
 75                self._connection_info = self._fetch_connection_info()
 76
 77            self._source_id = self._connection_info.source_id
 78
 79        return cast("str", self._source_id)
 80
 81    @property
 82    def source(self) -> CloudSource:
 83        """Get the source object."""
 84        if self._cloud_source_object:
 85            return self._cloud_source_object
 86
 87        self._cloud_source_object = CloudSource(
 88            workspace=self.workspace,
 89            connector_id=self.source_id,
 90        )
 91        return self._cloud_source_object
 92
 93    @property
 94    def destination_id(self) -> str:
 95        """The ID of the destination."""
 96        if not self._destination_id:
 97            if not self._connection_info:
 98                self._connection_info = self._fetch_connection_info()
 99
100            self._destination_id = self._connection_info.source_id
101
102        return cast("str", self._destination_id)
103
104    @property
105    def destination(self) -> CloudDestination:
106        """Get the destination object."""
107        if self._cloud_destination_object:
108            return self._cloud_destination_object
109
110        self._cloud_destination_object = CloudDestination(
111            workspace=self.workspace,
112            connector_id=self.destination_id,
113        )
114        return self._cloud_destination_object
115
116    @property
117    def stream_names(self) -> list[str]:
118        """The stream names."""
119        if not self._connection_info:
120            self._connection_info = self._fetch_connection_info()
121
122        return [stream.name for stream in self._connection_info.configurations.streams or []]
123
124    @property
125    def table_prefix(self) -> str:
126        """The table prefix."""
127        if not self._connection_info:
128            self._connection_info = self._fetch_connection_info()
129
130        return self._connection_info.prefix or ""
131
132    @property
133    def connection_url(self) -> str | None:
134        """The URL to the connection."""
135        return f"{self.workspace.workspace_url}/connections/{self.connection_id}"
136
137    @property
138    def job_history_url(self) -> str | None:
139        """The URL to the job history for the connection."""
140        return f"{self.connection_url}/job-history"
141
142    # Run Sync
143
144    def run_sync(
145        self,
146        *,
147        wait: bool = True,
148        wait_timeout: int = 300,
149    ) -> SyncResult:
150        """Run a sync."""
151        connection_response = api_util.run_connection(
152            connection_id=self.connection_id,
153            api_root=self.workspace.api_root,
154            workspace_id=self.workspace.workspace_id,
155            client_id=self.workspace.client_id,
156            client_secret=self.workspace.client_secret,
157        )
158        sync_result = SyncResult(
159            workspace=self.workspace,
160            connection=self,
161            job_id=connection_response.job_id,
162        )
163
164        if wait:
165            sync_result.wait_for_completion(
166                wait_timeout=wait_timeout,
167                raise_failure=True,
168                raise_timeout=True,
169            )
170
171        return sync_result
172
173    # Logs
174
175    def get_previous_sync_logs(
176        self,
177        *,
178        limit: int = 10,
179    ) -> list[SyncResult]:
180        """Get the previous sync logs for a connection."""
181        sync_logs: list[JobResponse] = api_util.get_job_logs(
182            connection_id=self.connection_id,
183            api_root=self.workspace.api_root,
184            workspace_id=self.workspace.workspace_id,
185            limit=limit,
186            client_id=self.workspace.client_id,
187            client_secret=self.workspace.client_secret,
188        )
189        return [
190            SyncResult(
191                workspace=self.workspace,
192                connection=self,
193                job_id=sync_log.job_id,
194                _latest_job_info=sync_log,
195            )
196            for sync_log in sync_logs
197        ]
198
199    def get_sync_result(
200        self,
201        job_id: int | None = None,
202    ) -> SyncResult | None:
203        """Get the sync result for the connection.
204
205        If `job_id` is not provided, the most recent sync job will be used.
206
207        Returns `None` if job_id is omitted and no previous jobs are found.
208        """
209        if job_id is None:
210            # Get the most recent sync job
211            results = self.get_previous_sync_logs(
212                limit=1,
213            )
214            if results:
215                return results[0]
216
217            return None
218
219        # Get the sync job by ID (lazy loaded)
220        return SyncResult(
221            workspace=self.workspace,
222            connection=self,
223            job_id=job_id,
224        )
225
226    # Deletions
227
228    def permanently_delete(
229        self,
230        *,
231        cascade_delete_source: bool = False,
232        cascade_delete_destination: bool = False,
233    ) -> None:
234        """Delete the connection.
235
236        Args:
237            cascade_delete_source: Whether to also delete the source.
238            cascade_delete_destination: Whether to also delete the destination.
239        """
240        self.workspace.permanently_delete_connection(self)
241
242        if cascade_delete_source:
243            self.workspace.permanently_delete_source(self.source_id)
244
245        if cascade_delete_destination:
246            self.workspace.permanently_delete_destination(self.destination_id)