# A stateless streaming application

In this tutorial we'll show how to write an application that implements a typical data enrichment pattern.

The application will consume `mediawiki.page_change.v1` from the public [eventstreams](https://stream.wikimedia.org/v2/ui/#/) service via an SSE connector. It 
will produce enriched events (with a fixed `content_body`) into a local filesystem sink.

To follow along you'll need Docker and Python (>=3.9) virtual environment. 

```bash
python3.9 -m venv venv
source venv/bin/activate
```

Before getting started, we'll need to install cookiecutter.
```
pip install cookiecutter
```

## 1. Scaffold a new application

```
cookiecutter https://gitlab.wikimedia.org/repos/data-engineering/eventutilities-python/ --directory cookiecutter-event-application
```

Or if you have the eventutilities-python repo cloned locally:
```
cookiecutter ${HOME}/repos/data-engineering/eventutilities-python/ --directory cookiecutter-event-application
```

For this application, we'll use default settings
```
  [1/4] project_name (event_platform_app): 
  [2/4] description (): 
  [3/4] eventutilities_python_version (0.15): 
  [4/4] uses_http (N): 
```

## 2. Configure the application

Next we'll create some local stream configuration, that will allow us to work with streams not yet available 
via the global EventStreamConfig service.
We'll declare `mediawiki.page_change.v1` (our source), which contains events with `mediawiki/page/change` schema.
We'll also declare `event_platform_app.sink` that also uses the `mediawiki/page/change` schema.

```
{
  "mediawiki.page_change.v1": {
    "schema_title": "mediawiki/page/change"
  },
  "event_platform_app.sink": {
    "schema_title": "mediawiki/page/change"
  },
  "event_platform_app.error": {
    "schema_title": "error"
  }
}
```

Now that have declared them, we can use these streams to configure source and sink connectors in `config.yaml`.

We'll consume data from a SSE stream:
```
  source:
    connector: sse
    stream: mediawiki.page_change.v1:1.1.0
```

Here we pinned version `1.1.0` of `mediawiki/page/change` to the stream.

We'll produce events into a local directory:
```
sink:
    connector: file
    stream: event_platform_app.sink:1.1.0
    options:
      uri: /tmp/out
```

Finally, we'll route errors into a local error sink:
```
error_sink:
    connector: file 
    stream: event_platform_app.error:2.1.0
    options:
      uri: /tmp/out
```

Dcumentation for `stream_manager` config settings can be found at  [https://doc.wikimedia.org/data-engineering/eventutilities-python/configuration.html](https://doc.wikimedia.org/data-engineering/eventutilities-python/configuration.html).

## 3. Implement an enrichment function

Finally we can implement an `enrich` function with our desired logic. Modify the application template under `event_platform_app/app.py`.

Here we modified the sample `enrich` to enrich `page_change` events with a content body.
```python
def enrich(event: dict) -> dict:
        event["revision"]['content_slots']['main']['content_body'] = "content body"
        return event
```

We'll need to also adjust the stream partition key to map the new schema:
```python
stream.partition_by(lambda e: (e["wiki_id"], e["page"]["page_id"]))
```
In this case `(wiki_id, page_id)` matches the partition scheme used internally (Kafka).


**Exercise**: update the test fixtures under `tests/events.json` and the `test_app.py` suite
to reflect this behaviour change.

## 4. Run the application locally

To run the application locally, you'll need `eventutilities-python[provided]` dependencies.

Install the application dependencies with
```
pip install --extra-index-url https://gitlab.wikimedia.org/api/v4/projects/1014/packages/pypi/simple eventutilities-python[provided]
```

and run the application with:
```
python event_platform_app/app.py --config config.yaml
```

After startup you'll `/tmp/out` will be populated with enriched events.
```
tail -f /tmp/out/.*
```

## 5. Run the application with docker

`.pipeline/blubber.yaml` will create images atop the base ones we use in k8s. It's good practice to work and test
with images before moving to the deployment phase.
These image package `eventutilites-python` provided deps.

Build the  image and run a `event_platform_app-devel` container with:
```
DOCKER_BUILDKIT=1 docker build --target development -t event_platform_app-devel -f .pipeline/blubber.yaml . 
docker run event_platform_app-devel
```

By default, events will be produced in `/tmp/out` inside the running container. You'll need to access the container
to inspect the application output.

## 6. Enrichment with external systems

The example we gave so far is limited to enriching a stream with some constant, self-contained, data transformation. In practice, 
application will often need to query external systems. `eventutilities_python` comes some helper for processing functions that
need to access HTTP endpoints.

By convention, such function must have the following signature:
```
my_func(event:dict, http_session: requests.Session) -> dict
```
and must be decorated with `eventutilities_python.functions.http_process_function`.

`request.Session` is provided by the `requests` package.


For example:
```python
from eventutilities_python.functions import http_process_function
 @http_process_function()
 def my_func(event: dict, http_session: requests.Session):
    event["enriched"] = http_session.get(...)["enriched_data"]
    return event
```

When dispatched via `stream_manager`, the `http_process_function` decorator will monkey patch
the session object with a `requests.Session` instance that can be scheduled on an async thread pool, 
and supports retry on error logic with backoff.
More details can be found at [https://doc.wikimedia.org/data-engineering/eventutilities-python/DESIGN.html#asynchronous-http-callback-via-datastream-minibatching](https://doc.wikimedia.org/data-engineering/eventutilities-python/DESIGN.html#asynchronous-http-callback-via-datastream-minibatching)/