# python-owsclient
OwsClient is a library for OWS RestAPI requests.

OwsClient uses [httpx](https://www.python-httpx.org/) which is a fully featured HTTP client for Python 3,
provides sync and async APIs, and support for both HTTP/1.1 and HTTP/2.

Install with:

```shell
pip install 'owsclient'
```

Usage:

```python
from owsclient.constants import QA_ENVIRONMENT
from owsclient import OwsClient

client = OwsClient(environment=QA_ENVIRONMENT, service_name="ows-dmp")

# Making synchronous request
response = client.get("ows-permissions", path="/hello/")
```

Using in dev and test environments:

```python
import os

from owsclient import OwsClient

from yourapp.context import get_correlation_id


os.environ["OWSREQUEST_SERVICE_MAP"] = "{\"ows-permissions\": \"http://localhost:5005/\"}"

client = OwsClient(
    environment="dev",
    service_name="ows-dmp",
    m2m_token_manager=...,  # set `M2MTokenManager` instance to add `Authorization: Bearer {token}` header
    get_correlation_id_func=get_correlation_id,
)

# Making synchronous request
response = client.get("ows-permissions", path="/hello/")
```

### Asynchronous Ows Client

Usage:

```python
from owsclient.constants import QA_ENVIRONMENT
from owsclient import AsyncOwsClient
from yourapp.context import get_correlation_id


client = AsyncOwsClient(
    environment=QA_ENVIRONMENT,
    service_name="ows-dmp",
    m2m_token_manager=...,  # set `AsyncM2MTokenManager` to add `Authorization: Bearer {token}` header,
    get_correlation_id_func=get_correlation_id,
)

# Making synchronous request
response = await client.get("ows-permissions", path="/hello/")

# Close transport and proxies.
await client.close()
```

### Impersonation Ows Client

Use [ImpersonationOwsClient](owsclient/impersonation_client.py) to make requests while impersonating another identity. This client requires an 
[ImpersonationM2MTokenManager](owsclient/m2m/impersonation.py) that can obtain tokens with impersonation claims.


```python
from owsclient import (
    ImpersonationM2MTokenManager,
    ImpersonationOwsClient
)

# Initialize the impersonation token manager
m2m_token_manager = ImpersonationM2MTokenManager(
    secrets_manager=secrets_manager,
    environment="qa",
    service_name="my-app",
)

# Create the impersonation client
client = ImpersonationOwsClient(
    environment="qa",
    service_name="my-app",
    m2m_token_manager=m2m_token_manager,
)

# Make a request while impersonating another identity
response = client.post(
    "ows-pdp",
    path="/identity/self/allowed-tenants/",
    json={
        "action": "view",
        "resource_type": "audience",
    },
    impersonated_identity_uuid="4d5f24f5-83f9-4989-9f82-0924a5feaf88",
)
```

Key Features:
- **Impersonation Support**: All HTTP methods (get, post, put, patch, delete) accept an `impersonated_identity_uuid` parameter.
- **Token Management**: Automatically obtains and includes impersonation tokens in the Authorization header.
- **Tolerance for Auth0 Instability**: Automatically retries with exponential backoff when generating an impersonation JWT token from Auth0 when their service has 502, 503, 504 status code responses
- **Same Interface**: Maintains the familiar OwsClient interface with added impersonation capability. (excluding `request_context_getter` support found in other OwsClient classes)

Requirements:
- The `m2m_token_manager` must be an instance of `ImpersonationM2MTokenManager`. Using a regular `M2MTokenManager` will raise a TypeError.

### POST/PUT JSON Request Body

When making requests with a JSON request body, be sure to pass a [JSON-Encoded data](https://www.python-httpx.org/quickstart/#sending-json-encoded-data) to `json` keyword argument. For example, `Decimal` is not JSON serializable and the following example will result in `TypeError: Object of type Decimal is not JSON serializable`.

#### DON'T
```python
ledger_entry = {
    "amount": Decimal(1439.23),
    "id": 1,
}

response = client.post("ows-hypothetical-ledger", path="/ledger/", json=ledger_entry)
# or for async:
response = await async_client.post("ows-hypothetical-ledger", path="/ledger/", json=ledger_entry)
```

Instead, explicitly ensure the `json` payload used reflects what the microservice can support/expects. `python-owsclient` does not do this automatically to avoid making incorrect assumptions about microservice behavior.

#### DO
```python
from simplejson import json  # You can use a different package
...

ledger_entry = {
    "amount": Decimal(1439.23),
    "id": 1,
}
# `allow_nan` avoids supporting out-of-range float values
# which this example's `ows-hypothetical-ledger` microservice cannot handle
json_ledger_entry = json.loads(json.dumps(ledger_entry, allow_nan=False))

response = client.post("ows-hypothetical-ledger", path="/ledger/", json=json_ledger_entry)
# or for async:
response = await async_client.post("ows-hypothetical-ledger", path="/ledger/", json=json_ledger_entry)

```

If you are trying to maintain consistency / porting from using `python-owsrequest` to use `python-owsclient` to make/send requests to another microservice, you will want to use something like `simplejson` and `allow_nan=False`.


### Service Discovery

python-owsclient supports service discovery using the following inputs:

- `environment`: `prod`, `qa`, `uat`, `test`, or `dev`
- `service_name`: The name of the target microservice (e.g., `ows-permissions`, `ows-account`)
- `OWSREQUEST_SERVICE_MAP` environment variable (optional): A stringified JSON dictionary mapping service names to URLs

Example of `OWSREQUEST_SERVICE_MAP`:
```
OWSREQUEST_SERVICE_MAP="{\"ows-permissions\": \"https://qa-ows-permissions.theorchard.io\", \"ows-account\": \"http://localhost:8888\", \"ows-participant\":\"https://qa-ows-participant.theorchard.io\"}"
```

#### Service Discovery Behavior

The `discover_service_url()` function in [services.py](owsclient/services.py) determines the base URL for each service based on the environment:

**For `prod` and `qa` environments:**
- Always constructs URLs using the pattern: `https://{environment}-{service_name}.theorchard.io`
- Ignores `OWSREQUEST_SERVICE_MAP` completely
- Examples:
  - `prod` + `ows-account` → `https://prod-ows-account.theorchard.io`
  - `qa` + `ows-permissions` → `https://qa-ows-permissions.theorchard.io`

**For `uat` environment:**
- First checks if the service is in `OWSREQUEST_SERVICE_MAP`
- If found, uses the mapped URL (can be localhost or any custom URL)
- If not found, constructs: `https://uat-{service_name}.theorchard.io`

**For `dev` and `test` environments:**
- First checks if the service is in `OWSREQUEST_SERVICE_MAP`
- If found, uses the mapped URL
- If not found, defaults to `qa` environment URL: `https://qa-{service_name}.theorchard.io`

**For any other environment value:**
- Always defaults to `qa` environment URL: `https://qa-{service_name}.theorchard.io`

#### URL Preparation

After service discovery, the `prepare_url()` method in [base.py](owsclient/base.py) combines the base URL with the request path:

1. Calls `discover_service_url()` to get the base service URL
2. Validates the URL scheme (must be `http` or `https`)
   - If no scheme is present, defaults to `http://`
3. Uses `urljoin()` to combine the base URL with the request path
4. Returns the complete request URL

Example:
```python
# Service discovery returns: http://localhost:8888
# Request path: /hello/
# Final URL: http://localhost:8888/hello/
```



### Testing Ows Client

OwsClient using [respx](https://lundberg.github.io/respx/) for mocking out the HTTPX, and HTTP Core, libraries.

Pytest usage:

```python
import httpx

from owsclient import OwsClient
from owsclient.test import OwsClientMock

ows_client = OwsClient(environment="test", service_name="ows-test")


def test_ows_permissions_hello(ows_client_mock: OwsClientMock) -> None:
    response_status_code = 200
    response_json = {"status": "ok"}

    ows_client_mock.get("ows-permissions", path="/hello/").mock(
        return_value=httpx.Response(status_code=response_status_code, json=response_json)
    )

    response = ows_client.get("ows-permissions", path="/hello/")

    assert response.status_code == response_status_code
    assert response.json() == response_json
```

This will also work for asynchronous Ows Client:

```python
import httpx
import pytest

from owsclient import AsyncOwsClient
from owsclient.test import OwsClientMock

client = AsyncOwsClient(environment="test", service_name="ows-test")


@pytest.mark.asyncio
async def test_ows_permissions_hello(ows_client_mock: OwsClientMock) -> None:
    response_status_code = 200
    response_json = {"status": "ok"}

    ows_client_mock.get("ows-permissions", path="/hello/").mock(
        return_value=httpx.Response(status_code=response_status_code, json=response_json)
    )

    response = await client.get("ows-permissions", path="/hello/")

    assert response.status_code == response_status_code
    assert response.json() == response_json
```

#### Asynchronous Ows Client with FastAPI

You can use asynchronous Ows client more efficiently with FastAPI asynchronous endpoint aggregating results from the given coroutines/futures.

Usage:

```python
import asyncio
from typing import Any

from fastapi import FastAPI

from owsclient.constants import QA_ENVIRONMENT
from owsclient import AsyncOwsClient

client = AsyncOwsClient(environment=QA_ENVIRONMENT, service_name="ows-dmp")

app = FastAPI(on_shutdown=[client.close])


@app.get("/hello/")
async def hello() -> Any:
    users_response, permissions_response = await asyncio.gather(
        client.get("ows-users", path="/hello/"),
        client.get("ows-permissions", path="/hello/")
    )
    return {
        "users": users_response.json(),
        "permissions": users_response.json(),
    }
```

#### Automatic Patching of Global OwsClient Instances

The `ows_client_mock` fixture automatically patches global `OwsClient` and `AsyncOwsClient` 
instances in your app modules. This is particularly useful for Python lambdas that instantiate 
clients as global variables.

When the fixture is used, it:
- Scans your modules for global `OwsClient` and `AsyncOwsClient` instances
- Replaces their `m2m_token_manager` with a mock token manager
- Automatically restores the original token managers after the test completes

This prevents "Failed to get M2M token" warnings during testing without requiring you to manually patch
the client for every unit test that uses `ows_client_mock`. 

Example:

```python
from owsclient import OwsClient
from owsclient.m2m.base import M2MTokenManager

# Global client (common pattern in Python lambdas)
m2m_token_manager = M2MTokenManager(...)
ows_client = OwsClient(
    environment="prod",
    service_name="my-app",
    m2m_token_manager=m2m_token_manager
)

def test_my_function(ows_client_mock: OwsClientMock) -> None:
    # The global `ows_client` is automatically patched
    ows_client_mock.get("ows-permissions", path="/test/").mock(
        return_value=httpx.Response(status_code=200, json={"status": "ok"})
    )
    
    # Your test code using the global ows_client
    # The mock token manager is automatically used

    # To verify the mock token manager was called
    ows_client_mock.mock_m2m_token_manager.get_token_string.assert_called_once()  # type: ignore[attr-defined]
```

For async:
```python
import pytest
from owsclient import AsyncOwsClient
from owsclient.m2m.base import AsyncM2MTokenManager

# Global async client
async_m2m_token_manager = AsyncM2MTokenManager(...)
async_ows_client = AsyncOwsClient(
    environment="prod",
    service_name="my-app",
    m2m_token_manager=async_m2m_token_manager
)

@pytest.mark.anyio
async def test_my_async_function(ows_client_mock: OwsClientMock) -> None:
    # The global `async_ows_client` is automatically patched
    ows_client_mock.get("ows-permissions", path="/test/").mock(
        return_value=httpx.Response(status_code=200, json={"status": "ok"})
    )

    # Your test code using the global async_ows_client
    response = await async_ows_client.get("ows-permissions", path="/test/")

    # To verify the mock async token manager was called
    ows_client_mock.mock_async_m2m_token_manager.get_token_string.assert_called_once()  # type: ignore[attr-defined]
```

For impersonation clients:
```python
import pytest
from owsclient import ImpersonationOwsClient
from owsclient.m2m.impersonation import ImpersonationM2MTokenManager

# Global impersonation client
impersonation_m2m_token_manager = ImpersonationM2MTokenManager(...)
impersonation_ows_client = ImpersonationOwsClient(
    environment="prod",
    service_name="my-app",
    m2m_token_manager=impersonation_m2m_token_manager
)

def test_my_impersonation_function(ows_client_mock: OwsClientMock) -> None:
    # The global `impersonation_ows_client` is automatically patched
    ows_client_mock.get("ows-permissions", path="/test/").mock(
        return_value=httpx.Response(status_code=200, json={"status": "ok"})
    )

    # Your test code using the global impersonation_ows_client
    impersonated_identity_uuid = "f94b0c5a-b520-486b-ac17-e59e9888b8bd"
    response = impersonation_ows_client.get(
        "ows-permissions", 
        path="/test/",
        impersonated_identity_uuid=impersonated_identity_uuid
    )

    # To verify the mock impersonation token manager was called with the correct UUID
    ows_client_mock.mock_impersonation_m2m_token_manager.get_token_string.assert_called_once_with(  # type: ignore[attr-defined]
        impersonated_identity_uuid=impersonated_identity_uuid
    )

##### Disabling Automatic Patching

You can disable the automatic patching behavior for specific tests using the 
`@pytest.mark.ows_client_mock_disable_m2m_patch` marker. This is useful when you need to
test the actual token manager behavior:

```python
import pytest
from owsclient.test import OwsClientMock

@pytest.mark.ows_client_mock_disable_m2m_patch
def test_with_real_token_manager(ows_client_mock: OwsClientMock) -> None:
    # The global ows_client will NOT be patched
    # It will use its original m2m_token_manager
    pass
```

### M2M Token Manager

Use M2MTokenManager to fetch m2m jwt token.

Usage:

```python
from owsclient import M2MTokenManager

secrets_manager = SecretsManager(region_name="us-east-1")

m2m_token_manager = M2MTokenManager(
    secrets_manager=secrets_manager,
    environment="qa",
    service_name="my-app",
    leeway_seconds=60
)

# Get token
token_string = m2m_token_manager.get_token_string("mysecret")
```

#### leeway parameter
Use the optional `leeway` parameter to ensure that `M2MTokenManager` fetches a new token before it expires. For example, 
setting `leeway=60` will prompt `M2MTokenManager` to retrieve a new token when it is due to expire within one minute.
The default leeway is `60` seconds if unset by the client.

**Never set leeway to a negative number.**. The token manager will attempt to use an expired JWT.

#### Cache Behavior

By default, M2MTokenManager will try to use an in-memory dictionary to retrieve the token string first. If the dictionary does not contain an unexpired token string, M2MTokenManager will fetch from AWS Secrets Manager and store the secret in the cache before returning the token string.

You can use a cache other than an in-memory dictionary by passing an object that implements the `Cache` protocol. For example, implementing `Cache` using `fakeredis` and using it with M2MTokenManager:

```python
class WannabeCache:
    """WannabeCache is not production-ready whatsoever and for illustrative purposes only."""

    def __init__(self) -> None:
        """Initialize fakeredis."""
        self._cache = fakeredis.FakeRedis()

    def get(self, key: str) -> Any:
        """Get key value from cache."""
        return self._cache.get(key)

    def set(self, key: str, value: Any, *, timeout: int | None = None) -> bool | None:
        """Set key with value to cache."""
        self._cache.set(key, value)
        return True

cache = WannabeCache()

m2m_token_manager = M2MTokenManager(
    secrets_manager=secrets_manager,
    environment="qa",
    service_name="my-app",
    cache=cache,
)
```

Suppose your machine's `M2MTokenManager` is sharing a cache with another machine's `M2MTokenManager`. In this case, you should override the `cache_key` when instantiating the `M2MTokenManager`.:

```python
m2m_token_manager = M2MTokenManager(
    secrets_manager=secrets_manager,
    environment="qa",
    service_name="my-app",
    cache=cache,
    cache_key="customCacheKeyOfYourChoice",
)
```

### Async M2M Token Manager

Use AsyncM2MTokenManager to fetch m2m jwt token from async code.

Usage:

```python
from owsclient import AsyncM2MTokenManager

secrets_manager = SecretsManager(region_name="us-east-1")

m2m_token_manager = AsyncM2MTokenManager(
    secrets_manager=secrets_manager,
    environment="qa",
    service_name="my-app",
)

# Get token
token_string = await m2m_token_manager.get_token_string()
```

### Impersonation M2M Token Manager

Use [ImpersonationM2MTokenManager](owsclient/m2m/impersonation.py) to obtain M2M JWT tokens with impersonation claims. 
This token manager fetches Auth0 client credentials from AWS Secrets Manager and generates tokens that allow your machine 
to impersonate another identity.

```python
from owsclient.m2m.impersonation import ImpersonationM2MTokenManager

# Initialize the token manager
m2m_token_manager = ImpersonationM2MTokenManager(
    secrets_manager=secrets_manager,
    environment="qa",
    service_name="my-app",
    leeway_seconds=60,
)

# Get a token for impersonating a specific identity
token = m2m_token_manager.get_token_string(
    impersonated_identity_uuid="4d5f24f5-83f9-4989-9f82-0924a5feaf88"
)
```

Key Features:
- Credential Caching: Caches Auth0 client credentials to reduce calls to AWS Secrets Manager
- Token Caching: Caches generated tokens per impersonated identity to avoid unnecessary Auth0 API calls

#### AsyncCache Behavior

By default, AsyncM2MTokenManager will try to use an in-memory dictionary to retrieve the token string first. If the dictionary does not contain an unexpired token string, AsyncM2MTokenManager will fetch from AWS Secrets Manager and store the secret in the cache before returning the token string.

You can use a cache other than an in-memory dictionary by passing an object that implements the `AsyncCache` protocol. For example, implementing `AsyncCache` using `fakeredis` and using it with AsyncM2MTokenManager:

```python
class WannabeAsyncCache:
    """WannabeAsyncCache is not production-ready whatsoever and for illustrative purposes only."""

    def __init__(self) -> None:
        """Initialize fakeredis."""
        self._cache = fakeredis.FakeRedis()

    async def get(self, key: str) -> Any:
        """Get key value from cache."""
        return self._cache.get(key)

    async def set(
        self, key: str, value: Any, ttl: Any = None, **kwargs: Any
    ) -> bool | None:
        """Set key with value to cache."""
        self._cache.set(key, value)
        return True

cache = WannabeAsyncCache()

m2m_token_manager = AsyncM2MTokenManager(
    secrets_manager=secrets_manager,
    environment="qa",
    service_name="my-app",
    cache=cache,
)
```

Suppose your machine's `AsyncM2MTokenManager` is sharing a cache with another machine's `AsyncM2MTokenManager`. In this case, you should override the `cache_key` when instantiating the `AsyncM2MTokenManager`.:

```python
m2m_token_manager = AsyncM2MTokenManager(
    secrets_manager=secrets_manager,
    environment="qa",
    service_name="my-app",
    cache=cache,
    cache_key="customCacheKeyOfYourChoice",
)
```