# Transformation service

## Make commands
Consider `make help` for supported commands:
```shell script
$ make help
Application: dapd_transformation_service

Run command:
  make <target>


  Local commands:
    venv                     create local virtual env (base.pip dependencies)
    testenv                  create local virtual env (test.pip dependencies)
    devenv                   create local virtual env (dev.pip dependencies)
    test                     run tests
    fmt                      reformat code with yapf
    fmtcheck                 dry run isort & yapf
    lint                     run pylint check
    typecheck                run mypy check
    build                    build package
    upload                   upload package on PyPI
    clean                    delete unnecessary files

  Docker commands:
    docker/login             login to AWS docker, aws configuration need to be set first
    docker/test/image        pull image used for tests execution
    docker/<local target>    universal docker wrapper for local targets
    docker/pg/start          start test pg instance
    docker/pg/stop           shutdown test pg instance
    docker/integration       run integration tests

  help                       this message
```

## Implementation notes:

- There are two parts of the Transformation service: API and Worker;
- API is managed by Gunicorn, Worker is started in separate process;
- IPC between API and Worker is based on ZeroMQ;
- Worker acts as a ZeroMQ server, API tends to be a client;
- Worker consists of two threads: the first thread is used for data reading from Kinesis,
the second one is used for ZeroMQ server;
- Synchronization between threads is based on `threading.Event`, so the thread with ZeroMQ server can manage another thread;
- Both API and Worker can be configured via environment variables;
- According to the `KINESIS_STREAM_NAME` option Worker decides which DSP and dimension entity
should be used, available options are `apple_music|spotify` and `dim_track|dim_playlist|dim_artist|dim_album`;
- There are two strategies for Kinesis reading: if there is at least one record for corresponding
dimension and DSP its `updated_at` value will be used for start reading from (`AT_TIMESTAMP` iterator option),
otherwise a Kinesis stream will be read from the beginning (`TRIM_HORIZON` iterator option);
- API supports incoming data validation based on JSON schemas;

## API examples:

#### Health check
```
curl -XGET 'http://localhost:9083/api/v1/health' -i
HTTP/1.0 200 OK
content-type: application/json
content-length: 16
Server: Werkzeug/1.0.1 Python/3.7.7
Date: Fri, 30 Oct 2020 13:38:04 GMT

{"status": "ok"}%
```
Available statuses: `ok`, `db_down`, `timeout`, `invalid_response`.

#### Fetching config
```
curl -XGET 'http://localhost:9083/api/v1/config' -i
HTTP/1.0 200 OK
content-type: application/json
content-length: 141
Server: Werkzeug/1.0.1 Python/3.7.7
Date: Fri, 30 Oct 2020 13:49:47 GMT

{"db_url": "postgresql+psycopg2://localhost/etldb", "env": "dev", "kinesis_stream_name": "", "zmq_server_host": "*", "zmq_server_port": 5555}%
```

#### Getting metrics
```
curl -XGET 'http://localhost:9083/api/v1/metrics' -i
HTTP/1.0 200 OK
content-type: application/json
content-length: 101
Server: Werkzeug/1.0.1 Python/3.7.7
Date: Fri, 30 Oct 2020 13:50:54 GMT

{"started_at": "2020-10-30T13:37:27.695076+00:00", "consumed_records": 0, "uptime": "0:13:27.218279"}%
```

#### Getting consumer state
```
curl -XGET 'http://localhost:9083/api/v1/consumer/' -i
HTTP/1.0 200 OK
content-type: application/json
content-length: 20
Server: Werkzeug/1.0.1 Python/3.7.7
Date: Fri, 30 Oct 2020 13:52:53 GMT

{"state": "enabled"}%
```
Avaliable states are: `enabled`, `disabled` and `unknown`.

#### Updating consumer state
```
curl -XPOST 'http://localhost:9083/api/v1/consumer/' -d '{"state": "enabled"}' -i
HTTP/1.0 204 No Content
Server: Werkzeug/1.0.1 Python/3.7.7
Date: Fri, 30 Oct 2020 13:53:32 GMT
```
```
curl -XPOST 'http://localhost:9083/api/v1/consumer/' -d '{"state": "disbled"}' -i
HTTP/1.0 204 No Content
Server: Werkzeug/1.0.1 Python/3.7.7
Date: Fri, 30 Oct 2020 13:53:32 GMT
```


## Developers guide

Use `pre-commit` for git hooks managing (https://pre-commit.com/#installation). Hook configuration
can be found at .pre-commit-config.yaml. `isort` and `yapf` tools are included to run as a part
of precommit hook.

To register a new hook run:
```shell script
pre-commit install
```

To run hooks without code committing:
```shell script
pre-commit run
```

Sometimes these tools can have a conflict regarding import sorting. In this case consider to
enable `include_trailing_comma` option at `setup.cfg` file.

Don’t forget to run `pylint` and `mypy` before code committing manually. It will save you from
wasting time at reading CI logs for failed build.

In order to check everything is fine (and have a dry run for formatting and import sorting)
consider following commands, e.g. for `dapd_transformation_service`:
```shell script
cd dapd_transformation_service
workon dapd_transformation_service # or './venv/bin/activate'
isort --check-only -rc . && yapf --recursive --verbose --parallel --diff . \
&& pylint --rcfile=pylintrc ./${PWD##*/} ./tests && mypy ${PWD##*/} && echo 'Exit code:' $?
```

It will run all checks inside a current directory.


Run in Docker

  make docker/image/build

  docker run \
    -e ENVIRONMENT=stage -e AWS_DEFAULT_REGION=us-east-1 \
    -v ~/.aws/:/app/.aws/ -v `pwd`/env:/dapd_transformation_service/env \
    -it dapd_transformation_service:DAPD-1473