Metadata-Version: 2.4
Name: northflank
Version: 1.0.0
Summary: Python SDK for the Northflank platform API
Author-email: Northflank <contact@northflank.com>
License-Expression: MIT
Project-URL: Homepage, https://northflank.com
Project-URL: Documentation, https://northflank.com/docs/v1/api
Requires-Python: <3.14,>=3.9
Description-Content-Type: text/markdown
Requires-Dist: httpx>=0.27
Requires-Dist: websockets>=13.0
Requires-Dist: typing_extensions>=4.0
Provides-Extra: generator
Requires-Dist: black>=24.0; extra == "generator"
Provides-Extra: dev
Requires-Dist: pytest>=7.0; extra == "dev"
Requires-Dist: black>=24.0; extra == "dev"
Requires-Dist: httpx>=0.27; extra == "dev"
Requires-Dist: websockets>=12.0; extra == "dev"
Requires-Dist: basedpyright>=1.30; extra == "dev"

# Northflank Python SDK

The official Python client for the [Northflank](https://northflank.com) API.
It provides typed endpoint methods for every Northflank resource — services,
sandboxes, GPU workloads, jobs, addons, secrets, volumes, domains, metrics, 
and more — plus exec and logs (range fetch and live tailing).

## Installation

```sh
pip install northflank
```

Requires Python 3.9–3.13.

## Quick start

```python
from northflank import ApiClient

client = ApiClient(api_token="your-api-token")

# List the services in a project
services = client.list.services(project_id="my-project")
for service in services.data["services"]:
    print(service["name"])
```

Every call returns an `ApiCallResponse` with `.data`, `.status`, and `.error`.
Method arguments are keyword-only and fully typed — request bodies and
response `data` are described by generated `TypedDict`s, so editors and type
checkers can validate payloads.

## Error handling

By default the client **raises** `ApiCallError` on any non-2xx response —
this matches the convention of `requests`, `httpx`, `boto3`, and most Python
SDKs:

```python
from northflank import ApiCallError

try:
    client.get.service(project_id="my-project", service_id="does-not-exist")
except ApiCallError as exc:
    print(exc.status, exc.message)
```

Pass `throw_on_http_error=False` to instead surface the error on the response
object and never raise:

```python
client = ApiClient(api_token="...", throw_on_http_error=False)

resp = client.get.service(project_id="my-project", service_id="does-not-exist")
if resp.error is not None:
    print(resp.error.status, resp.error.message)
```

## Working with services

### Create a service

```python
client.create.service.deployment(
    project_id="my-project",
    data={
        "name": "my-service",
        "billing": {"deploymentPlan": "nf-compute-20"},
        "deployment": {
            "instances": 1,
            "external": {"imagePath": "nginx:latest"},
            "docker": {"configType": "default"},
        },
        "ports": [
            {"name": "http", "internalPort": 80, "public": True, "protocol": "HTTP"},
        ],
    },
)
```

### Wait for it to be running

```python
import time

while True:
    service = client.get.service(project_id="my-project", service_id="my-service")
    status = service.data["status"]["deployment"]["status"]
    if status == "COMPLETED":
        break
    if status == "FAILED":
        raise RuntimeError("deployment failed")
    time.sleep(3)
```

### Pause, resume, and delete

```python
client.pause.service(project_id="my-project", service_id="my-service")
client.resume.service(project_id="my-project", service_id="my-service", data={})
client.delete.service(project_id="my-project", service_id="my-service")
```

## Exec

### One-shot commands

`run_service_command` runs a command, waits for it to finish, and returns an
`ExecResult` (`exit_code`, `stdout`, `stderr`, `status`):

```python
result = client.exec.run_service_command(
    project_id="my-project",
    service_id="my-service",
    command="cat /etc/os-release",
)
print(result.exit_code, result.stdout)

# Commands that need a shell (pipes, &&, redirection) — pass `shell`:
client.exec.run_service_command(
    project_id="my-project",
    service_id="my-service",
    command="echo hello && uname -a",
    shell="bash -c",
)
```

The same one-shot API works for jobs and addons:

```python
client.exec.run_job_command(project_id="my-project", job_id="my-job", command="ls")
client.exec.run_addon_command(
    project_id="my-project", addon_id="my-redis", command="redis-cli ping", shell="bash -c"
)
```

### Interactive sessions

`open_service_session` opens a persistent connection: write to the process's
stdin, iterate output as it streams, resize the terminal. Use it as a context
manager so the WebSocket is always closed. Send input, **read the output it
produces, then `send_eof`** — closing stdin immediately after a `send` can
race the proxy.

```python
with client.exec.open_service_session(
    project_id="my-project", service_id="my-service", command="cat"
) as session:
    session.resize(rows=40, columns=120)
    session.send("hello\n")
    for chunk in session:            # chunk.stream is "stdout" / "stderr"
        print(chunk.data, end="")
        if "hello" in chunk.data:
            break
    session.send_eof()               # close stdin
    result = session.wait()          # ExecResult once the process exits
```

`open_job_session` and `open_addon_session` cover jobs and addons. The async
client exposes `aopen_service_session` / `aopen_job_session` /
`aopen_addon_session`, returning an `AsyncExecSession` driven with
`async with` and `async for`.

## Logs

### Fetch a recent range

```python
lines = client.logs.fetch_service_logs(
    project_id="my-project",
    service_id="my-service",
    line_limit=100,
    text_includes="error",
)
for line in lines:
    print(line.ts, line.log)
```

### Live tail

`tail_service_logs` opens a WebSocket and returns a `LogTail` you iterate for
`LogLine` items as they arrive:

```python
with client.logs.tail_service_logs(
    project_id="my-project", service_id="my-service", recv_timeout=30.0
) as tail:
    for line in tail:
        print(line.ts, line.log)
```

`tail_job_logs` and `tail_addon_logs` cover jobs and addons; the async client
provides `atail_service_logs` / `atail_job_logs` / `atail_addon_logs`
(`AsyncLogTail`, driven with `async for`).

## Metrics

`client.get.{service,job,addon}.metrics` fetches resource metrics. `query_type`
selects a time range or a single snapshot, `metric_types` takes a list, and
time-range arguments accept native `datetime` objects:

```python
from datetime import datetime, timedelta, timezone
from northflank import METRIC_TYPES

metrics = client.get.service.metrics(
    project_id="my-project",
    service_id="my-service",
    query_type="range",
    metric_types=["cpu", "memory"],
    start_time=datetime.now(timezone.utc) - timedelta(hours=1),
)
print(metrics.data)
```

`MetricType` (a `Literal` of every valid metric) and `METRIC_TYPES` (all of
them, as a tuple) are exported from `northflank` for annotation and reuse.

## Async usage

`AsyncApiClient` mirrors `ApiClient` with awaitable calls:

```python
import asyncio
from northflank import AsyncApiClient

async def main():
    client = AsyncApiClient(api_token="your-api-token")
    services = await client.list.services(project_id="my-project")
    print(services.data)

asyncio.run(main())
```

## Pagination

Paginated list endpoints return one page by default. Use `.all()` to drain
every page into a single response, or `.pagination.get_next_page()` to step
through them:

```python
# All pages at once
every_service = client.list.services.all(project_id="my-project")

# One page at a time
page = client.list.services(project_id="my-project")
while page.pagination and page.pagination.has_next_page:
    page = page.pagination.get_next_page()
```

## Configuration

`ApiClient` reads configuration from arguments or environment variables:

| Environment variable      | Purpose                                  |
| ------------------------- | ---------------------------------------- |
| `NF_API_TOKEN`            | API token (also `NORTHFLANK_API_TOKEN`)  |
| `NF_API_HOST` / `NF_HOST` | API base URL (default `https://api.northflank.com`) |

```python
# Explicit
client = ApiClient(api_token="...", base_url="https://api.northflank.com")

# From the environment
client = ApiClient()
```

## More examples

The [`examples/`](examples/) directory has runnable scripts covering
sandboxes, interactive exec sessions, log fetching and tailing, GPU workloads,
networking, persistent volumes, and parallel sandbox pools. Each one reads
`NF_API_TOKEN` and `NF_PROJECT_ID` from the environment.
