airbytehq / airbytehq/PyAirbyte

Feature Proposal: `CloudWorkspace.get_custom_source()`

Ouverte
#574 3 commentaires 0 réactions 0 personnes assignées Voir sur GitHub
accepting pull requests
Langage dominant
Python
Étoiles
344
Forks
77
Merge moyen
1 j 11 h
PR mergées (30 j)
35

Description

## Challenge

As of today, it's a manual process to test custom connectors in PyAirbyte. Whenever the source definition changes, there's a manual step needed to copy-paste the yaml into the runtime environment where you are running PyAirbyte.

## Proposal

Add to `CloudWorkspace`:
- `CloudWorkspace.custom_sources`: `list[CustomSourceConnector]` - a list of custom sources defined in this workspace.
- `CloudWorkspace.get_custom_source(*, definition_id: str | None, name: str) -> CustomSourceConnector` - Allow getting custom connector by ID or by name. User should provide one or the other, but not both.

## Pseudocode

`CustomSourceConnector` might have a definition something like this:

```py
@dataclass
class CustomSourceConnector:
workspace: CloudWorkspace
name: str
manifest: dict
version: int | None
definition_id: str

def as_local_source(
self,
/,
config: dict | None = None,
config_overlay: dict | None = None,
) -> Source:
"""Return a local source object which can be executed locally.
If `config` is provided, it will replace the Cloud config.
If `config_overlay` is provided, it will be overlayed on top of the Cloud config.

Note: By design, PyAirbyte cannot retrieve secrets from the Cloud API endpoints.
Any secret config parameters will be returned as `******` and will need to be
replaced using a config overlay.
"""
...
```

### Usage Example A

We pass the manifest to `get_source()`.

```py
import airbyte as ab

my_workspace = ab.CloudWorkspace(
client_id=...,
client_secret=...,
)

my_source_definition = my_workspace.get_custom_source(name="My Test")
my_source = ab.get_source(
"source-my-test",
declarative_manifest=my_source_definition.manifest,
config={...},
)

# Now we can work on it like a normal local source...
my_source.check()
my_source.read()
```

### Usage Example B

We use the `as_local_source()` method to get a local `Source` object.

```py
import airbyte as ab

my_workspace = ab.CloudWorkspace(
client_id=...,
client_secret=...,
)

my_source_definition = my_workspace.get_custom_source(name="My Test")
my_source = my_source_definition.as_local_source(config_overlay={"password": ...})

# Now we can work on it like a normal local source...
my_source.check()
my_source.read()
```

Guide de contribution

Ouvrir le guide de contribution

Évaluation

Cette issue n'a pas encore été évaluée.

Recevez les nouvelles issues par e-mail

Un résumé court des issues GitHub adaptées aux débutants.