shopcloud-eventbus

Shopcloud-EventBus

Async Processing, write to Log an Event and send it to a Queue

graph TD;
    AppEngine-->Log;
    Log-->EventBus;
    EventBus-->AppEngine;

Usage

$ pip install shopcloud-eventbus

$ eventbus init
$ eventbus deploy

Vanilla:

from shopcloud_eventbus import Event

event = Event(
    name="de.talk-point.platform/module/model/sync",
    model=self,
)
event.add_task(
    queue="default",
    url=f"module/api/model/{self.id}/action/",
    json={}
)
event.fire()

Django

use for example the shopcloud-django-toolbox library

from shopcloud_django_toolbox import Event

event = Event(
    name="de.talk-point.platform/module/model/sync",
    model=self,
)
event.add_task(
    queue="default",
    url=f"module/api/model/{self.id}/action/",
    json={}
)
event.fire()

Install

Services

$ gcloud services enable cloudtasks.googleapis.com --project="test-eventbus-project"
$ gcloud pubsub topics create events --project='test-eventbus-project'
$ gcloud tasks queues create default --project='test-eventbus-project' --location='europe-west3'

Log Sink

Create a log sink with the following filter resource.type="gae_app" "event-fire"

$ eventbus --help
$ eventbus deploy

Development

$ python3 setup.py install
$ python3 -m shopcloud_eventbus -d init --project="test-eventbus-project" --region="europe-west3" --app-endpoint="https://test-eventbus-project.ey.r.appspot.com" --secrethub-endpoint-user="talk-point/app-eventbus-test/production/evenbus-user" --secrethub-endpoint-pwd="talk-point/app-eventbus-test/production/evenbus-pwd"
$ python3 -m shopcloud_eventbus -d deploy

dann kann das Script getestet werden per

$ cd .eventbus
$ python3 -m venv venv
$ source venv/bin/activate
$ pip install -r requirements.txt
$ functions-framework --target main_http --debug

oder deployen und eine Test-Message durchsenden

Deploy

$ rm -rf build dist
$ pip3 install wheel twine
$ python3 setup.py sdist bdist_wheel
$ twine upload dist/*