Metadata-Version: 2.4
Name: reachcollective-utils
Version: 0.1.38
Summary: Utilities for developing RESTful API, Open API projects with FastAPI and Python
Author: Juan Minaya
Author-email: jminaya@reachcollective.com
Requires-Python: >=3.12
Classifier: License :: OSI Approved :: MIT License
Requires-Dist: fastapi[standard] (>=0.120.0,<0.163.0)
Requires-Dist: opentelemetry-api (>=1.40.0,<2.0.0)
Requires-Dist: opentelemetry-instrumentation-logging (>=0.61b0,<1.0)
Requires-Dist: opentelemetry-sdk (>=1.40.0,<2.0.0)
Requires-Dist: sqlmodel (>=0.0.27,<0.0.28)
Requires-Dist: structlog (>=25.4.0,<26.0.0)
Description-Content-Type: text/markdown

# ReachCollective / APIs / Utils
Utilities for developing RESTful API, Open API projects with FastAPI and Python

# Requirements
- sqlmodel (>=0.0.27,<0.0.28)
- fastapi[standard] (>=0.120.0,<0.163.0)

# Installation

### Poetry
```
poetry add reachcollective-utils
```
### Pip
```
pip install reachcollective-utils
```

### OpenTelemetry logging

OpenTelemetry support and its required dependencies are included with the
standard package installation.

Select the provider once when the application starts:

```python
from reachcollective.utils.log import AppLog

AppLog.configure(provider='otel')
```

Existing logging calls remain unchanged and are emitted as OpenTelemetry JSON
records to `stdout`:

```python
AppLog.log(
    'error',
    'Connection error 1',
    service='paypal',
    retry_count=5,
)
```

#### Trace correlation

There are two possible configurations for `AppLog`.

##### OpenTelemetry logging enabled

When the backend is running with OpenTelemetry tracing, enable OpenTelemetry
logging once during application startup, before any calls to `AppLog.log()`:

```python
from fastapi import FastAPI
from reachcollective.utils.log import AppLog


def create_app() -> FastAPI:
    AppLog.configure(provider='otel')
    return FastAPI()
```

If `AppLog.log()` runs while an OpenTelemetry span is active—for example,
inside an instrumented FastAPI request—the resulting JSON record includes the
same `trace_id` and `span_id` as that span. This allows a log to be correlated
with its distributed trace:

```json
{
  "body": "Connection error 1",
  "severity_text": "ERROR",
  "attributes": {
    "service": "paypal",
    "retry_count": 5
  },
  "trace_id": "0x542d8a9c55c6523d62cf9a9cb837c2e8",
  "span_id": "0x417e83d5c50604d3"
}
```

`AppLog` does not create traces or spans. They must be created by the
backend's OpenTelemetry SDK, automatic instrumentation, or manual
instrumentation. A log emitted during startup, in an uninstrumented background
task, or anywhere else without an active span has zero-valued `trace_id` and
`span_id`. Configuring the OTel logging provider changes the log output, but by
itself does not enable tracing.

##### OpenTelemetry logging not enabled

If the application does not call `AppLog.configure(provider='otel')`, `AppLog`
continues to use `AppAWSLog` by default:

```python
from reachcollective.utils.log import AppLog


AppLog.log(
    'critical',
    'Validation error 4',
    service='onfido',
)
```

The resulting compact JSON keeps the AWS log format and does not contain
OpenTelemetry `trace_id` or `span_id` fields, even if tracing is enabled
elsewhere in the backend:

```json
{
  "service": "onfido",
  "event": "Validation error 4",
  "timestamp": "2026-08-25T02:41:13.389667Z",
  "level": "critical"
}
```

# Components

## Handling Errors
Copy and paste this fragment into main.py of your FastAPI project.
```python
# Handling Errors
# If an APIException, HTTPException is not defined in the router, the following line is executed
app.add_exception_handler(ValidationError, APIRender.pydantic_validation_handler) # type: ignore  # code: 422
app.add_exception_handler(RequestValidationError, APIRender.request_validation_handler) # type: ignore  # code: 422 
app.add_exception_handler(APIException, APIRender.api_exception_handler) # type: ignore
app.add_exception_handler(ValueError, APIRender.value_error_handler) # type: ignore # code: 400 
app.add_exception_handler(HTTPException, APIRender.http_exception_handler) # type: ignore
```

## DataGrid
List and paginate data from a model, accepting filters, sorting, and relationships.

### 1. Modify models
Add `to_dict()` function to each SQL Model, `to_dict()` gets relationships.

```python

class Post(SQLModel, table=True):
    __tablename__ = "posts"

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    name: str | None
    status: str | None
    comments: list["Comment"] | None = Relationship(back_populates="post", sa_relationship_kwargs={"lazy": "raise"})

    def to_dict(self, *args, **kwargs):
        data = super().model_dump(*args, **kwargs)
        if 'comments' in self.__dict__:
            data['comments'] = []
            if self.quota_groups:
                data['comments'] = [quota_group.to_dict(*args, **kwargs) for quota_group in self.quota_groups]

        return data


class Comment(SQLModel, table=True):
    __tablename__ = "comments"

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    description: str | None
    post: Post | None = Relationship(back_populates="comments", sa_relationship_kwargs={"lazy": "raise"})

    def to_dict(self, *args, **kwargs):
        data = super().model_dump(*args, **kwargs)
        if 'post' in self.__dict__:
            data['post'] = {}
            if self.survey:
                data['post'] = self.survey.model_dump(*args, **kwargs)

        return data
 ```

### 2. Basic use
```python
# URL: /posts?sort=-created&filter[status]=active&view=paginate


from app.models import Post
from reachcollective.utils.datagrid import DataGrid

@router.get('/posts')
async def list_(
    request: Request,
    db: AsyncSession = Depends(get_session)
):
    return await DataGrid(db, Post, request).init().get()
```

### 2. Custom filters
```python
# URL: /posts?sort=-created&filter[status]=active&filter[name]=demo&view=paginate


from app.models import Post
from reachcollective.utils.datagrid import DataGrid

@router.get('/posts')
async def list_(
    request: Request,
    db: AsyncSession = Depends(get_session)
):
    datagrid = DataGrid(db, Post, request)
    datagrid.qp.filters.personalize = ['name']
    datagrid.init()

    for key, value in datagrid.params['filters']['customize'].items():
        match key:
            case 'name':
                datagrid.qb.stmt = datagrid.qb.stmt.where(Post.name.ilike(f'%{value}%'))

    return await datagrid.get()
 ```

### 3. Relationships
```python
# URL: /posts?sort=-created&filter[status]=active&view=paginate&with=comments


from app.models import Post
from reachcollective.utils.datagrid import DataGrid

@router.get('/posts')
async def list_(
    request: Request,
    db: AsyncSession = Depends(get_session)
):
    return await DataGrid(db, Post, request).init().get()
```

### 4. Results
```json
{
  "current_page": 1,
  "data": [
    {
      "id": "18a6ac49-cb2f-41c7-ac14-ff5360210c65",
      "name": "demo",
      "status": "active"
    },
    {
      "id": "81f08423-288f-4e74-bcbb-edf5ef37f1fe",
      "name": "demo",
      "status": "active"
    },
    {
      "id": "a017c74f-f0bb-4f65-9faf-36581812bbe7",
      "name": "Test 4",
      "status": "active"
    }
  ],
  "total": 683,
  "per_page": 15,
  "total_pages": 46
}
```

```json
{
  "current_page": 1,
  "data": [
    {
      "id": "18a6ac49-cb2f-41c7-ac14-ff5360210c65",
      "name": "demo",
      "status": "active",
      "comments": [
        {
          "id": "28a6ac49-cb2f-41c7-ac14-ff5360210c15",
          "name": "Hello"
        },
        {
          "id": "38a6ac49-cb2f-41c7-ac14-ff5360210c25",
          "name": "World"
        }
      ]
    },
    {
      "id": "81f08423-288f-4e74-bcbb-edf5ef37f1fe",
      "name": "demo",
      "status": "active",
      "comments": []
    },
    {
      "id": "a017c74f-f0bb-4f65-9faf-36581812bbe7",
      "name": "Test 4",
      "status": "active",
      "comments": [
        {
          "id": "38a6ac49-cb2f-41c7-ac14-ff5360210c35",
          "name": "Hello"
        }
      ]
    }
  ],
  "total": 683,
  "per_page": 15,
  "total_pages": 46
}
```

## Query Params
Parses, sanitizes and formats data from a URL query params. NOTE: Does not require `sqlmodel`

```python
# URL: /query-params?sort=-last_updated&with=survey,profile&size=2&filter[name]=young&filter[status]=active|deactivate&filter[survey_id]=994c231c|8a900c77&view=paginate

from reachcollective.utils.datagrid import QueryParams
from reachcollective.utils import APIRender

@router.get('/query-params')
async def test_query_params(request: Request):
    try:
        return QueryParams(request).get()
    except HTTPException as e:
        return APIRender.error(e.detail, e.status_code)
```

### Results
```json
{
  "filters": {
    "init": {},
    "equals": {
      "name": "young",
      "status": [
        "active",
        "deactivate"
      ],
      "survey_id": [
        "994c231c",
        "8a900c77"
      ]
    },
    "customize": {}
  },
  "with": [
    "survey",
    "profile"
  ],
  "sort": [
    {
      "col": "last_updated",
      "dir": "desc",
      "by": "-last_updated"
    }
  ],
  "view": "paginate",
  "page": 1,
  "size": 2
}
```

## APIRender
Centralizes fastAPI and third-party responses into a single class. Used in FastaPI routes. NOTE: Require `pydantic`

```python
from reachcollective.utils import APIRender

@router.get('/api-render')
async def test_query_params(request: Request):
    try:
        # TODO: Your code here
    except HTTPException as e:
        return APIRender.error(e.detail, e.status_code)
```

