abs-pipelines-core 0.1.1
Core utilities for building MongoDB/Cosmos DB aggregation pipelines
License: MIT
Requires Python: <4.0,>=3.11

# abs-pipelines-core

MongoDB aggregation pipeline builders and file operations for record enrichment.

## Installation

```bash
# From PyPI
pip install abs-pipelines-core

# Or with Poetry
poetry add abs-pipelines-core
```

## Quick Start - Record Enrichment

Use `prepare_enrichment()` to get everything you need in one call:

```python
from abs_pipelines_core import RecordEnricher

enricher = RecordEnricher(
    blob_service=blob_service,
    user_collection_name="user_documents"
)

# Fetch ALL fields for the entity
fields = (await field_repository.get_all(filters)).get("founds", [])

# Prepare enrichment config (with optional record filter)
config = enricher.prepare_enrichment(
    fields=fields,
    record_ids=["rec_id_1", "rec_id_2"],  # Optional: filter by specific IDs
    option_field_ids=["status_field_id"]   # Optional: fields needing label resolution
)

# Apply to query
find = ListFilter()
if config.pre_filter:
    find.pre_filters = config.pre_filter
find.reference_fields = AggregationStage(pipeline=config.pipeline)

# Get records
response = await repository.get_all(find, collection_name)

# Add file URLs (post-processing)
response = await enricher.process_file_fields(config.file_field_ids, response)
```

## File Operations - EntityRecordBlobHandler

File upload/download operations for entity records:

```python
from abs_pipelines_core import EntityRecordBlobHandler

# Initialize with blob repository
blob_handler = EntityRecordBlobHandler(blob_repository)

# Upload single file
result = await blob_handler.upload_record_field_file(entity_id, field_id, file)
# Returns: {field_id: file_id}

# Upload multiple files (filename format: {field_id}_actualfilename)
results = await blob_handler.upload_record_field_files(entity_id, files)
# Returns: [{field_id: file_id}, ...]

# Get file URL
url = await blob_handler.get_record_field_file(file_id)

# Get multiple file URLs
urls = await blob_handler.get_record_field_files(entity_id, file_ids)
# Returns: [{file_id: url}, ...]

# Delete file
await blob_handler.delete_record_field_file(file_id)

# Delete field folder
await blob_handler.delete_record_field_folder(entity_id, field_id)

# Delete entity folder
await blob_handler.delete_record_folder(entity_id)
```

### Storage Path Convention

Files are stored with this path structure:
- `entity-{entity_id}/field-{field_id}/{file_id}`

## Fields with Associations Pipeline

Use `build_fields_with_associations_pipeline()` to fetch fields with their associated definitions populated:

```python
from abs_pipelines_core import build_fields_with_associations_pipeline

# Build the pipeline
pipeline = build_fields_with_associations_pipeline(user_collection_name="users_documents")

# Add a match stage and run
full_pipeline = [{"$match": {"entity_id": entity_id}}, *pipeline]
fields = await db.fields.aggregate(full_pipeline).to_list(1000)

# Result for association field:
# {
#     "_id": "field_123",
#     "type": "association",
#     "reference": {
#         "entity_id": "entity_456",
#         "entity_fields": [  # <-- Populated!
#             {"_id": "...", "name": "Name", "type": "text", ...},
#         ]
#     }
# }

# Result for user field:
# {
#     "_id": "field_789",
#     "type": "user",
#     "user_filter": {
#         "is_active": true,
#         "users": [  # <-- Populated!
#             {"_id": "...", "name": "John Doe", ...},
#         ]
#     }
# }
```

## EnrichmentConfig

The `prepare_enrichment()` method returns an `EnrichmentConfig` with:

| Property | Type | Description |
|----------|------|-------------|
| `pipeline` | `List[Dict]` | Combined `$lookup` and `$addFields` aggregation stages |
| `pre_filter` | `Optional[Dict]` | Filter schema for record_ids (use with `find.pre_filters`) |
| `file_field_ids` | `List[str]` | Field IDs for file URL processing |

## Individual Pipeline Methods

### `build_lookup_pipeline(fields)`

Build `$lookup` stages for association and user fields:

```python
lookup_pipeline = enricher.build_lookup_pipeline(fields)
```

### `build_option_label_pipeline(field_id, value_to_label, is_multi_select)`

Build `$addFields` stage for option label resolution:

```python
value_to_label = {"uuid-1": "Active", "uuid-2": "Inactive"}
option_pipeline = enricher.build_option_label_pipeline(
    field_id="status",
    value_to_label=value_to_label,
    is_multi_select=False
)
```

### `process_file_fields(file_fields, response, expiry_minutes)`

Attach file URLs to records (async, post-processing):

```python
response = await enricher.process_file_fields(
    file_fields=["file_field_id"],
    response=response,
    expiry_minutes=5
)
```

## Package Structure

```
abs_pipelines_core/
├── __init__.py          # Public API exports
├── record_enricher.py   # RecordEnricher, schemas, and protocols
├── blob_handler.py      # EntityRecordBlobHandler for file operations
└── fields_pipeline.py   # build_fields_with_associations_pipeline
```

## Exports

```python
from abs_pipelines_core import (
    # Record enrichment
    RecordEnricher,                          # Aggregation pipeline builder
    EnrichmentConfig,                        # Return type for prepare_enrichment()
    FieldConfig,                             # Schema for field configurations
    FieldReference,                          # Schema for association references
    OptionConfig,                            # Schema for select options
    BlobStorageProtocol,                     # Protocol for blob service (get_file_url)

    # Blob/file operations
    EntityRecordBlobHandler,                 # File operations handler
    RequestData,                             # Request data for file uploads
    BlobRepositoryProtocol,                  # Protocol for full blob repository

    # Fields pipeline
    build_fields_with_associations_pipeline, # Fields with associations aggregation
)
```

## Requirements

- Python 3.11+
- motor >= 3.6.0
- pymongo >= 4.10.0
- pydantic >= 2.10.0