Metadata-Version: 2.4
Name: rushabhdynamodb
Version: 0.1.2
Summary: Production-focused DynamoDB helper library for query orchestration, filtering, batch reads, and pandas conversion
Author: Rushabh Panchal
License-Expression: MIT
Project-URL: Homepage, https://github.com/Rushee123/rushabhdynamodb
Project-URL: Repository, https://github.com/Rushee123/rushabhdynamodb
Project-URL: Issues, https://github.com/Rushee123/rushabhdynamodb/issues
Project-URL: Documentation, https://github.com/Rushee123/rushabhdynamodb/blob/main/FUNCTION_USAGE_GUIDE.md
Keywords: dynamodb,aws,boto3,query,python
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
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.9
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: boto3<2,>=1.34
Requires-Dist: pandas<3,>=1.5
Provides-Extra: dev
Requires-Dist: build>=1.0; extra == "dev"
Requires-Dist: twine>=5.0; extra == "dev"
Dynamic: license-file

# rushabhdynamodb

`rushabhdynamodb` is a lightweight helper package for common DynamoDB workflows in Python projects.
It wraps recurring boto3 patterns for querying, scanning, filtering, batching, and DataFrame conversion.

## Why Use It

- Create consistently configured boto3 clients and resources with sensible defaults.
- Run key-based queries and scans through a single `fetch` interface.
- Compose nested filter conditions (`AND`, `OR`, `NOT`) with an easy dictionary structure.
- Retrieve large ranges safely with chunk-based fetch operations.
- Convert DynamoDB payloads into pandas DataFrames quickly.

## Installation

Install from PyPI:

```bash
pip install rushabhdynamodb
```

Install from TestPyPI while still resolving dependencies from PyPI:

```bash
pip install --index-url https://pypi.org/simple --extra-index-url https://test.pypi.org/simple rushabhdynamodb==0.1.1
```

## Quick Start

```python
from rushabhdynamodb import DynamoDBQueryHelper

helper = DynamoDBQueryHelper(table_name="sensor-data", region_name="ap-south-1")

rows = helper.fetch(
    partition_key="device_id",
    partition_value="DVC1001",
    sort_key="timestamp",
    sort_op=">=",
    sort_value="2026-01-01 00:00:00",
    limit=100,
)

print(f"Fetched {len(rows)} rows")
```

## API Overview

### AWSConnector

Use `AWSConnector` to build boto3 clients/resources with explicit settings or environment-based configuration.

```python
from rushabhdynamodb import AWSConnector

connector = AWSConnector(service_name="dynamodb", region_name="ap-south-1")
client = connector.get_client()
resource = connector.get_resource()
```

Supported environment variables:

- `AWS_REGION` or `AWS_DEFAULT_REGION`
- `AWS_ACCESS_KEY_ID`
- `AWS_SECRET_ACCESS_KEY`
- `AWS_SESSION_TOKEN`
- `AWS_MAX_POOL_CONNECTIONS`
- `AWS_CONNECT_TIMEOUT`
- `AWS_READ_TIMEOUT`
- `AWS_RETRY_MAX_ATTEMPTS`
- `AWS_RETRY_MODE`
- `AWS_TCP_KEEPALIVE`

### DynamoDBQueryHelper

`DynamoDBQueryHelper` exposes utility methods:

- `fetch(...)`
- `batch_get(...)`
- `fetch_in_time_chunks(...)`
- `dynamo_to_dataframe(items)`
- `parse_time_delta(time_str)`
- `insert_to_dynamodb_for_alldata(df)`

## Fetch Filters

Simple condition example:

```python
{"field": "status", "op": "=", "value": "active"}
```

Nested logical filter example:

```python
{
    "logical": "AND",
    "conditions": [
        {"field": "status", "op": "=", "value": "active"},
        {"field": "temperature", "op": ">=", "value": 20},
    ],
}
```

Supported operators:

- `=`
- `!=`
- `<`
- `<=`
- `>`
- `>=`
- `between`
- `in`
- `not in`
- `like`
- `not like`

## Documentation

Full usage details and query examples are also available in [FUNCTION_USAGE_GUIDE.md](https://github.com/Rushee123/rushabhdynamodb/blob/main/FUNCTION_USAGE_GUIDE.md). The complete guide is reproduced below.

### Create an AWSConnector

Import:

```python
from rushabhdynamodb import AWSConnector
```

#### Option A: Environment-based configuration

```python
connector = AWSConnector(service_name="dynamodb")
client = connector.get_client()
resource = connector.get_resource()
```

Required region source:

- `AWS_REGION` or `AWS_DEFAULT_REGION`

Optional credentials:

- `AWS_ACCESS_KEY_ID`
- `AWS_SECRET_ACCESS_KEY`
- `AWS_SESSION_TOKEN`

Optional client tuning:

- `AWS_MAX_POOL_CONNECTIONS`
- `AWS_CONNECT_TIMEOUT`
- `AWS_READ_TIMEOUT`
- `AWS_RETRY_MAX_ATTEMPTS`
- `AWS_RETRY_MODE`
- `AWS_TCP_KEEPALIVE`

#### Option B: Direct values in code

```python
connector = AWSConnector(
    service_name="dynamodb",
    region_name="ap-south-1",
    aws_access_key_id="YOUR_ACCESS_KEY",
    aws_secret_access_key="YOUR_SECRET_KEY",
    aws_session_token="YOUR_SESSION_TOKEN",
    max_pool_connections=200,
    connect_timeout=8,
    read_timeout=20,
    retry_max_attempts=5,
    retry_mode="standard",
    tcp_keepalive=True,
)
```

### Create a DynamoDBQueryHelper

```python
from rushabhdynamodb import DynamoDBQueryHelper

helper = DynamoDBQueryHelper(
    table_name="sensor-data",
    region_name="ap-south-1",
)
```

### API Reference

#### AWSConnector(service_name, region_name=None, ...)

- `service_name: str`
- `region_name: Optional[str]`
- `aws_access_key_id: Optional[str]`
- `aws_secret_access_key: Optional[str]`
- `aws_session_token: Optional[str]`
- `max_pool_connections: Optional[int]`
- `connect_timeout: Optional[int]`
- `read_timeout: Optional[int]`
- `retry_max_attempts: Optional[int]`
- `retry_mode: Optional[str]`
- `tcp_keepalive: Optional[bool]`

Methods:

- `get_client()`
- `get_resource()`

#### DynamoDBQueryHelper.fetch(...)

- `partition_key: Optional[str]`
- `partition_value: Any`
- `sort_key: Optional[str]`
- `sort_op: str = "="`
- `sort_value: Any`
- `filters: Optional[Dict[str, Any]]`
- `limit: Optional[int]`
- `descending: bool = False`
- `projection: Optional[Iterable[str]]`
- `index_name: Optional[str]`

Returns: `List[Dict[str, Any]]`

#### DynamoDBQueryHelper.batch_get(keys, projection=None)

- `keys: List[Dict[str, Any]]`
- `projection: Optional[Iterable[str]]`

Returns: `List[Dict[str, Any]]`

#### DynamoDBQueryHelper.fetch_in_time_chunks(...)

- `start_timestamp: str` in format `YYYY-MM-DD HH:MM:SS`
- `partition_key: str`
- `partition_value: Any`
- `sort_key: str = "tagtime"`
- `chunk_hours: int = 2`
- `max_days: int = 30`
- `index_name: Optional[str] = None`
- `projection: Optional[Iterable[str]] = None`
- `descending: bool = False`
- `delay_every_n_chunks: int = 3`
- `delay_seconds: float = 0.1`
- `records_per_batch: int = 500`
- `batch_delay: float = 0.05`

Yields dictionaries with:

- `chunk_number`
- `items`
- `start_time`
- `end_time`
- `record_count`
- `batch_count`

#### DynamoDBQueryHelper.dynamo_to_dataframe(items)

- `items: List[Dict[str, Any]]`

Returns: `pandas.DataFrame`

#### DynamoDBQueryHelper.parse_time_delta(time_str)

Supported examples:

- `1-12:30:15`
- `36:15`
- `00:02:30`

Returns: `datetime.timedelta`

#### DynamoDBQueryHelper.insert_to_dynamodb_for_alldata(df)

Expected DataFrame columns:

- `tagid`
- `tagtime`
- `tagvalue`
- `mapped_tagid`
- `imputed`
- `tag_original_time`
- `grade`

### Filter Dictionary Format

Simple condition:

```python
{"field": "status", "op": "=", "value": "active"}
```

Nested condition:

```python
{
    "logical": "AND",
    "conditions": [
        {"field": "status", "op": "=", "value": "active"},
        {
            "logical": "OR",
            "conditions": [
                {"field": "temperature", "op": ">=", "value": 20},
                {"field": "grade", "op": "in", "value": ["A", "B"]},
            ],
        },
    ],
}
```

Supported logical operators: `AND`, `OR`, `NOT`

### Fetch Examples

Assume:

```python
helper = DynamoDBQueryHelper(table_name="sensor-data", region_name="ap-south-1")
```

#### Exact partition key

```python
items = helper.fetch(partition_key="device_id", partition_value="DVC1001")
```

#### Partition + sort range

```python
items = helper.fetch(
    partition_key="device_id",
    partition_value="DVC1001",
    sort_key="timestamp",
    sort_op="between",
    sort_value=["2026-03-01 00:00:00", "2026-03-02 00:00:00"],
)
```

#### Partition query with projection and limit

```python
items = helper.fetch(
    partition_key="device_id",
    partition_value="DVC1001",
    projection=["device_id", "timestamp", "temperature"],
    limit=25,
)
```

#### Query with filter

```python
items = helper.fetch(
    partition_key="device_id",
    partition_value="DVC1001",
    filters={"field": "status", "op": "=", "value": "active"},
)
```

#### Query through a GSI

```python
items = helper.fetch(
    partition_key="mapped_tagid",
    partition_value=101,
    sort_key="tagtime",
    sort_op=">=",
    sort_value="2026-03-01 00:00:00",
    index_name="mapped_tagid-tagtime-index",
)
```

### Notes and Best Practices

- Use `query` patterns (partition key present) whenever possible to reduce read cost.
- Keep projections small for better throughput and lower latency.
- Use `fetch_in_time_chunks` for large time windows to avoid throttling spikes.

## Development

```bash
python -m pip install -U build twine
python -m build
python -m twine check dist/*
```

## License

MIT
