Metadata-Version: 2.4
Name: prefect-couchbase
Version: 1.0.0
Summary: Prefect collection for connecting flows and tasks to Couchbase.
Author-email: Couchbase Examples <devadvocacy@couchbase.com>
License-Expression: Apache-2.0
Project-URL: Homepage, https://github.com/couchbase-examples/couchbase-prefect
Project-URL: Repository, https://github.com/couchbase-examples/couchbase-prefect
Project-URL: Issues, https://github.com/couchbase-examples/couchbase-prefect/issues
Keywords: prefect,couchbase,etl,database,connector
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Natural Language :: English
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Database
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: couchbase>=4.4.0
Requires-Dist: prefect>=3.0.0
Provides-Extra: dev
Requires-Dist: build>=1.2; extra == "dev"
Requires-Dist: coverage[toml]>=7.4; extra == "dev"
Requires-Dist: pytest>=8.0; extra == "dev"
Requires-Dist: pytest-cov>=5.0; extra == "dev"
Requires-Dist: ruff>=0.6; extra == "dev"
Dynamic: license-file

# prefect-couchbase

A Prefect collection for connecting Prefect flows and tasks to Couchbase through the official Couchbase Python SDK.

## What you can use it for

- Store Couchbase connection details as a Prefect block and load them safely from flows.
- Read, write, and delete documents from Couchbase buckets/scopes/collections in Prefect tasks.
- Run SQL++ queries from Prefect flows while reusing the same Couchbase SDK connection lifecycle.

This connector does **not** replace the Couchbase Python SDK. It provides Prefect-native credentials and a small convenience wrapper around the SDK so workflows can manage credentials, lifecycle, and common operations consistently.

## End-user quickstart

Install the package:

```bash
pip install prefect-couchbase
```

Create and save a credentials block:

```python
from prefect_couchbase import CouchbaseCredentials

credentials = CouchbaseCredentials(
    connection_string="couchbases://cb.example.cloud.couchbase.com",
    username="analytics_user",
    password="super-secret-password",
)
credentials.save("prod-couchbase", overwrite=True)
```

Use the block from a flow:

```python
from prefect import flow, task
from prefect_couchbase import CouchbaseCredentials


@task
def upsert_hotel(document_id: str, name: str) -> dict:
    credentials = CouchbaseCredentials.load("prod-couchbase")

    with credentials.get_connector(
        bucket="travel-sample", scope="_default", collection="_default"
    ) as couchbase:
        couchbase.upsert(document_id, {"type": "hotel", "name": name})
        return couchbase.get(document_id)


@flow
def demo() -> dict:
    return upsert_hotel("prefect::hotel::1", "Prefect Grand Hotel")


if __name__ == "__main__":
    print(demo())
```

Expected output:

```text
{'type': 'hotel', 'name': 'Prefect Grand Hotel'}
```

## Provision Couchbase

### Recommended: Couchbase Capella

Use Capella when Prefect Cloud workers or any hosted execution infrastructure need to reach Couchbase over the public internet.

1. Create a Capella cluster.
2. Create or choose a bucket, scope, and collection.
3. Add a database user with access to the bucket.
4. Add the worker's outbound IP address to Capella's allowed IP list.
5. Use the Capella SDK connection string, which starts with `couchbases://`.
6. Configure the Prefect block with the connection string, username, and password.

### Local or self-managed Couchbase

Use local/self-managed Couchbase when your Prefect worker runs on the same machine or private network.

1. Start Couchbase Server locally.
2. Create a bucket such as `travel-sample` or `prefect-demo`.
3. Create a user with bucket read/write permissions.
4. Use a connection string such as `couchbase://localhost`.

For a quick local evaluation, the Couchbase Python SDK examples and Docker documentation are the best reference for the currently supported server image and initialization flow.

## Configuration reference

`CouchbaseCredentials` fields:

| Field | Description |
| --- | --- |
| `connection_string` | Couchbase SDK connection string, such as `couchbase://localhost` or `couchbases://cb.example.cloud.couchbase.com`. |
| `username` / `password` | Password authentication credentials. Store these as Prefect secrets through the block. |
| `cert_path` | Optional trusted CA certificate path for password authentication; client certificate path for certificate authentication. |
| `key_path` / `trust_store_path` | Required with `cert_path` when using certificate authentication. |
| `profile` | Optional SDK config profile, for example `wan_development`. |
| `options` | Extra keyword arguments passed to `couchbase.options.ClusterOptions`. |

`CouchbaseConnector` methods:

- `get_bucket(bucket=None)`
- `get_scope(bucket=None, scope=None)`
- `get_collection(bucket=None, scope=None, collection=None)`
- `upsert(key, value, **kwargs)`
- `get(key, content_as=dict, **kwargs)`
- `remove(key, **kwargs)`
- `query(statement, *args, **kwargs)`
- `ping(**kwargs)`
- `close()`

## Tutorial: runnable local flow

1. Install dependencies:

   ```bash
   pip install prefect-couchbase
   ```

2. Export connection settings:

   ```bash
   export COUCHBASE_CONNECTION_STRING="couchbase://localhost"
   export COUCHBASE_USERNAME="Administrator"
   export COUCHBASE_PASSWORD="password"
   export COUCHBASE_BUCKET="travel-sample"
   ```

3. Save a credentials block once from the current environment:

   ```bash
   python examples/save_credentials_block.py
   ```

4. Run the example flow, which loads the saved block:

   ```bash
   python examples/couchbase_flow.py
   ```

5. Validate by checking that the flow prints:

   ```text
   Stored document: {'type': 'hotel', 'name': 'Prefect Grand Hotel'}
   ```

## Troubleshooting

- `Authentication failed`: verify username/password and bucket permissions.
- `Unambiguous timeout`: verify the worker can reach the cluster and that Capella allowed IPs include the worker.
- `bucket must be provided`: pass `bucket="..."` to `get_connector()` or to the individual operation.
- TLS errors with Capella: use a `couchbases://` connection string and ensure the cluster endpoint is correct.

## Developer documentation

### Local development

```bash
git clone https://github.com/couchbase-examples/couchbase-prefect.git
cd couchbase-prefect
uv sync --extra dev
```

Run tests and coverage:

```bash
uv run pytest
```

Run linting:

```bash
uv run ruff check .
uv run ruff format --check .
```

Build the package:

```bash
uv run python -m build
```

Run the example smoke check without a live Couchbase cluster:

```bash
uv run python -m py_compile examples/save_credentials_block.py examples/couchbase_flow.py
```

Run the examples against a live cluster by setting the environment variables from the tutorial, saving the block with `uv run python examples/save_credentials_block.py`, and then executing `uv run python examples/couchbase_flow.py`.

### Release and publishing path

1. Ensure `uv run pytest`, `uv run ruff check .`, and `uv run python -m build` pass.
2. Update the version tag according to the repository release policy.
3. Build distributions with `uv run python -m build`.
4. Publish to PyPI or the chosen private package index with `twine upload dist/*` after configuring maintainer credentials.
5. Register the collection with Prefect by installing the package in the worker environment. The package exposes the `prefect.collections` entry point `prefect_couchbase = prefect_couchbase`.

## Research and design notes

This implementation follows Prefect's collection pattern from official integrations such as `prefect-redis`, `prefect-snowflake`, and `prefect-sqlalchemy`: typed Prefect blocks hold credentials and user code imports a small package-specific public API. It also mirrors the Couchbase authentication and collection access shape used by the Couchbase Airflow provider: password authentication uses `PasswordAuthenticator`, certificate authentication uses `CertificateAuthenticator`, and helper methods resolve bucket/scope/collection objects.

## Known limitations

- The unit test suite uses SDK mocks so it can run without a live Couchbase service. Live connectivity should be validated in the environment where the Prefect worker runs.
- The connector intentionally exposes a minimal wrapper. Advanced SDK features should be accessed from `connector.cluster` or native SDK objects returned by `get_bucket`, `get_scope`, and `get_collection`.
