# abacus_common_logic

This library is designed to avoid using repeatable code and logic within tightly coupled Abacus services. Currently used by:
* [lambda-abacus](https://github.com/theorchard/lambda-abacus)
* [lambda-documents](https://github.com/theorchard/lambda-documents)
* [ows-ledger](https://github.com/theorchard/ows-ledger)
* [ows-payee](https://github.com/theorchard/ows-payee)
* [ows-payment](https://github.com/theorchard/ows-payment)
* [ows-royalties](https://github.com/theorchard/ows-royalties)
* [ows-abacus-contract](https://github.com/theorchard/ows-abacus-contract)
* [ows-abacus-event](https://github.com/theorchard/ows-abacus-event)
* [ows-abacus-state](https://github.com/theorchard/ows-abacus-state)
* [ows-abacus-worksheet](https://github.com/theorchard/ows-abacus-worksheet)
* [ows-abacus-schedule](https://github.com/theorchard/ows-abacus-schedule)
* [ows-abacus-account](https://github.com/theorchard/ows-abacus-account)
* [ows-abacus-legacy-sync](https://github.com/theorchard/ows-abacus-legacy-sync)

## Installation

```bash
# Using uv (recommended)
uv add abacus-common-logic

# Using pip
pip install -i https://pypi.theorchard.io/pypi/ abacus-common-logic
```

## Local Development

If you're making changes to the library and want to test locally before publishing:

### Prerequisites

Install [uv](https://docs.astral.sh/uv/) (modern Python package manager):
```bash
# macOS/Linux
curl -LsSf https://astral.sh/uv/install.sh | sh

# Or with Homebrew
brew install uv
```

### Setup Development Environment

```bash
# Clone the repository
git clone https://github.com/theorchard/python-abacus-common-logic.git
cd python-abacus-common-logic

# Install all dependencies and create virtual environment
uv sync

# This automatically:
# - Creates .venv/ directory
# - Installs the package in editable mode
# - Installs all dependencies from pyproject.toml
# - Configures private PyPI registry (pypi.theorchard.io)
```

### Testing

```bash
# Run linter
make lint

# Run tests
make test

# Run integration tests
make test_integration
```

#### Testing with Docker

You can run the tests inside Docker using the commands that are used in the pipeline.

You'll first need to login to Docker with the shared AWS account:

```bash

# Login to Docker
awsume <shared_profile>
make docker_login

# Run unit and functional tests
make ci_unit_lint

# Run integration tests
make ci_test_integration
```

### Watch mode / Editable mode

To test your local changes in a downstream service:

```bash
# In the downstream project
pip install -e /path/to/python-abacus-common-logic

# Or with uv
uv add --editable /path/to/python-abacus-common-logic
```

## Usage

### Marshalling

This library uses `flask_marshmallow` and provides us with following custom
fields, which can be used in Marshmallow schemas creation:

- Enum
- FormattedDate
- MoneyAmount
- NonemptyString (typo, Empty should be capitalized)
- NonNegativeNumber
- IntegerId
- Percentage  # numbers range (0, 100)

Example:

```python

from abacus_common_logic.marshalling.custom_fields import ma


class BaseSalesFileSchema(ma.Schema):
    """Base sales file schema."""

    accounting_period_id = ma.IntegerId()
    name = ma.NonemptyString(required=True)
    row_count = ma.NonNegativeInteger(required=True)
    master_url = ma.NonemptyString(required=True)
    staging_url = ma.NonemptyString(required=True)
    sales_file_status = ma.Enum(options=('NEW','IN_PROGRESS','DONE'), required=True)

```

### SQLAlchemy BaseModel

Use this model as a base class for your sqlalchemy models. It populates your model
with following additional fields:

- created_at = Datetime
- last_modified = Datetime
- created_by = String
- last_modified_by = String

and has a bunch of helper methods:

#### `find_by_name`
Returns first row from the table `WHERE name=YOUR_VALUE`

#### `get_by_id`
Returns a row, found by provided primary_key value.
`None` if not found.

#### `get_by_id_or_error`
Returns a row, found by provided primary_key value.
If not found, calls `flask.abort`.

#### `delete_by_id_or_error`
The same as for `get_by_id_or_error`, but if a row
is found, deletes it.

#### `default_order`
Returns a field, that would be used as a default one
for ordering your set of results.

#### `build`
Constructs a new ORM object, generates autoincrement,
but doesn't populate the database

#### `create`
The same as `build` and commits into the database.

#### `current_timestamp`
Returns correct date (by default for tz.tzutc())

#### `commit_changes`
Commits the session, adding any new objects if necessary

#### `update_attributes`
Updates your object's attributes with provided values.
Updates `last_modified_by` and `last_modified` fields
by default


## Utilities

Package with following helper modules:

### `dates`
Module with methods related to dates and time:

#### `safe_format_date`
Formats datetime object, uses `strftime`

#### `parse_date`
Parses a date from provided string, uses `strptime`

#### `current_timestamp`
Returns current time in provided timezone, UTC by default

#### `operating_date`
Returns provided timestamp in operating timezone.

Usage:
```python

from abacus_common_logic.utils.dates import parse_date, operating_date


timestamp = parse_date('2020-01-01', '%Y-%m-%d')
operating_timestamp = operating_date(timestamp)  # tweaks tz to Eastern Time

```

### `enum`
Receives a `name` and dict of `elements`. Returns a `namedtuple`.

Usage:

```python

from abacus_common_logic.utils.enum import enum

ABACUS_DOGS = enum(
    'AbacusDogs',
    THE_BEARD='Reggie',
    THE_CHONK='Henry',
    THE_FLOOF='Geno'
)

print (ABACUS_DOGS.THE_BEARD) # output: 'Reggie'
```

### `request`
Provides methods for handling requests, standardizing logic across services.

#### `BooleanFilter`

A utility enum for tri-state boolean filtering. Normalizes common truthy / falsy tokens (e.g. `"yes"`, `"0"`).

**Values**:
* **`TRUE`**
* **`FALSE`**
* **`ALL`** — Represents a disabled filter (maps to `None`)

**Class Methods**
* **`parse(value, default?)`**

**Methods**:
* **`to_bool()`** — converts the enum to a `bool`, or `None` if `BooleanFilter.ALL`

### `users`
Provides methods for flask to extract user info from request headers.
First it will check for profile headers,
then fall back to check for grass headers.

Most implementing services should add `set_flask_user_details_from_headers`
to flask's *before_request* hook like:
```python
from abacus_common_logic.utils.users import set_flask_user_details_from_headers
@app.before_request
def before_request_start():
    set_flask_user_details_from_headers()
```

### `authorization`
Provides functions to help with authorization checking.

Some Abacus endpoints are open to some profile types that should have access to only some accounts.
The `permissions_authorize_many_accounts` function is here to help with checking that the requester's profile has access to the requested accounts.

```python
from abacus_common_logic.utils.authorization import permissions_authorize_many_accounts


@contract_api.route('/contracts/account/dataloader', methods=['POST'])
def get_contracts_by_account_dataloader_handler():
    # Authorization checks ...

    account_access = permissions_authorize_many_accounts(
        ows_client,
        g.request_context.profile_type,
        g.request_context.profile_id,
        account_ids
    )

    if not account_access:
        return flaskify(response.create_error_response(
            code=ERROR_CODE_FORBIDDEN,
            message=ERROR_MESSAGE_FORBIDDEN_USER,
            status=403
        ))
```

This function only checks access for certain profile types.
The default list of profile types that are checked is stored in `PROFILE_TYPES_TO_CHECK`.
You can override this default list by passing the `profile_types_to_check` argument to either of these two functions.

```python
from abacus_common_logic.utils.authorization import permissions_authorize_many_accounts
from abacus_common_logic.utils.authorization import PROFILE_TYPES_TO_CHECK

# ...

account_access = permissions_authorize_many_accounts(
    ows_client,
    g.request_context.profile_type,
    g.request_context.profile_id,
    [account_id],
    profile_types_to_check=(PROFILE_TYPES_TO_CHECK + ['NewProfile'])
)
```


## Test Utils

Modules to use in unit/integration tests across services.

### `factories`
Base classes for creation own factories to populate database
with test values. Base usage:


```python

from abacus_common_logic.test_utils.factories import BaseModelFactoryWithTimestamps
import factory

class AccountingPeriodFactory(BaseModelFactoryWithTimestamps):
    """Factory to create accounting period models for testing."""

    class Meta:
        """Meta definition for the factory."""

        model = factory.SOME_MODEL_CLASS

    name = factory.Sequence(lambda n: f'Month 202{n}')
    status = 'SOME STATUS'
```

See detailed usage here [factory-boy](https://factoryboy.readthedocs.io/en/latest/).