Skip to main content

Library for separating data input, output and processing in your business application.

Project description


Build Status Maintainability Test Coverage PyPI version PyPI - Python Version

Library for separating data input, output and processing in your business application.


Disclaimer: the library is sooo pre-alpha.

Motivation & main idea

You have tons of business logic. You like clean architecture, but you're sane. You like dynamic structure of Python, but you're tired of runtime errors. You want to break things a little less and keep moving fast. You're is the right place.

Mario is a framework for business logic. Like Django or Flask for web-services.

It makes you put logic to pipelines: sets of pipes, each pipe does only one thing and only non-complex types can be transferred from pipe to pipe.

Each pipe is one of 3 types: input, output, processing. Input and output should be non-complex (like really non-complex, cyclomatic complexity ~3), processing pipes should be pure.


pip install super-mario


Here they are.

Usage example

Here is simple pipeline, that send notifications on new comments in Jira tickets to Slack.

class JiraCommentsNotificationPipeline(BasePipeline):
    pipeline = [

    def fetch_new_comments(jira_ticket_id: str) -> ImmutableContext:
        return {'new_comments':
                    hours=0, minutes=0, seconds=0, milliseconds=0,

    def fetch_users_mapping(new_comments: List[IssueComment]) -> ImmutableContext:
        return {
            'jira_to_slack_id_mapping': dict(User.objects.filter(
                jira_id__in=[c['user_id'] for c in new_comments],
            ).values_list('jira_id', 'slack_id'))

    def generate_slack_message(
        jira_ticket_id: str,
        new_comments: List[IssueComment],
        jira_to_slack_id_mapping: Mapping[str, str],
    ) -> ImmutableContext:
        message = '\n'.join([
            f'@{jira_to_slack_id_mapping[c["user_id"]]} wrote'
            f'comment for {jira_ticket_id}: "{c["text"]}"'
            for c in new_comments
        return {'message': message}

    def send_slack_message(message: str) -> None:

# run pipeline for specific ticket


We would love you to contribute to our project. It's simple:

  1. Create an issue with bug you found or proposal you have. Wait for approve from maintainer.
  2. Create a pull request. Make sure all checks are green.
  3. Fix review comments if any.
  4. Be awesome.

Here are useful tips:

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

super_mario-0.0.4.tar.gz (9.4 kB view hashes)

Uploaded source

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page