Python streaming system for building pipelines with REST APIs, scheduled tasks, and message queues (RabbitMQ, ZeroMQ, Kafka).

ANK lets you compose processors into chains and pipelines. Each processor is a link in the chain; connect them to form data flows.

Features:

• Pipeline and streaming processing
• REST API interface (Flask)
• Scheduled tasks (cron-style)
• Message queues: RabbitMQ, ZeroMQ, Kafka, Redis

• Python 3.10+

From PyPI:

pip install ank

From Docker (no local Python required):

# Build from source
git clone git@github.com:sunary/ank.git
cd ank
docker build -t ank .

# Or pull from Docker Hub (when published)
# docker pull sunary/ank

From source:

git clone git@github.com:sunary/ank.git
cd ank
pip install -e .

With ank installed (pip/Docker):

ank create TestService
# or with app type:
ank create TestService -c APIApp
cd TestService

With Docker only (no local install):

docker run --rm -v $(pwd):/workspace -w /workspace ank create TestService
cd TestService

App types: BaseApp (pipeline, default), APIApp (REST API), ScheduleApp (cron-based).

Edit processor.py:

from ank.components.pipe_app import PipeApp

class ExampleApp(PipeApp):

    def start(self):
        for i in range(100):
            self.chain_process({'content': i})

    def process(self, message=None):
        return message['content'] + 1

Edit services.yml to wire processors and dependencies:

services:
  StartApp:
    class: processor.StartApp
    arguments: [$Redis, '%batch_size%']

  Redis:
    class: redis.Redis
    arguments: ['%redis_host%', '%redis_port%']

  LogApp:
    class: ank.components.log_app.LogApp
    arguments: ~

chains:
  - StartApp
  - LogApp

Edit settings.yml with your parameters:

parameters:
  redis_host: localhost
  redis_port: 6379
  batch_size: 100

ank run -fs settings.yml

Or programmatically:

from ank.program_loader import main
main(file_setting='settings.yml')

For branching processors, return a message with a flags key to control routing:

def process(self, message=None):
    return {
        'content': result,
        'flags': [True, False]  # Process 1st branch only
    }

• [True, True] → process both branches
• [True, False] → process 1st branch only
• [False, True] → process 2nd branch only
• [False, False] → stop chain

services:
  ServiceName:
    class: module.ClassName
    arguments: [$OtherService, '%param%']  # ~ for no args

chains:
  - Service1
  - Service2
  - [BranchA, BranchB]  # branching

• $Service — inject another service
• %param% — value from settings.yml

Parameters referenced in services.yml are read from settings.yml. Use ank gen_setting to create a template.

See the examples/ directory:

• streaming_app/ — pipeline with Join/Split
• api_app/ — REST API
• schedule_app/ — scheduled tasks

Use ank without installing Python locally:

# Create new app (output in current directory)
docker run --rm -v $(pwd):/workspace -w /workspace ank create MyApp

# Run a service
docker run --rm -v $(pwd):/workspace -w /workspace ank run -fs settings.yml

# Generate settings
docker run --rm -v $(pwd):/workspace -w /workspace ank gen_setting -fs _setting.yml

From your project directory (after ank create):

ank build
docker run --entrypoint /bin/sh <image_id>

See LICENSE.