马里奥

用于在您的业务应用程序中分离数据输入、输出和处理的库。

免责声明：该库处于预_alpha阶段。

动机与主要思想

您有成堆的业务逻辑。您喜欢干净的架构，但您很理智。您喜欢Python的动态结构，但您厌倦了运行时错误。您希望减少一些错误并保持快速开发。这里是您的正确地方。

马里奥是一个用于业务逻辑的框架。类似于用于Web服务的Django或Flask。

它让您将逻辑放入管道：一系列管道，每个管道只做一件事，并且只能传输非复杂类型。

每个管道属于以下3种类型之一：输入、输出、处理。输入和输出应该是非复杂的（真正意义上的非复杂，圈复杂度约为3），处理管道应该是纯的。

安装

pip install super-mario

文档

它们在这里.

使用示例

这是一个简单的管道，用于将Jira工单中的新评论通知发送到Slack。

class JiraCommentsNotificationPipeline(BasePipeline):
    pipeline = [
        'fetch_new_comments',
        'fetch_users_mapping',
        'generate_slack_message',
        'send_slack_message',
    ]

    @input_pipe
    def fetch_new_comments(jira_ticket_id: str) -> ImmutableContext:
        return {'new_comments':
            fetch_jira_comments(
                ticket_id=jira_ticket_id,
                date_from=datetime.datetime.now().replace(
                    hours=0, minutes=0, seconds=0, milliseconds=0,
                ),
            ),
        }

    @input_pipe
    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'))
        }

    @process_pipe
    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}

    @output_pipe
    def send_slack_message(message: str) -> None:
        send_message(
            destination='slack',
            channel=COMMENTS_SLACK_CHANNEL_ID,
            text=message,
        )

# run pipeline for specific ticket
JiraCommentsNotificationPipeline().run(jira_ticket_id='TST-12')

贡献

我们非常欢迎您为我们的项目贡献力量。很简单

1. 创建一个问题，报告您发现的错误或您提出的建议。等待维护者的批准。
2. 创建一个拉取请求。确保所有检查都为绿色。
3. 如有必要，修复审查评论。
4. 做得好。

以下是一些有用的提示

• 您可以使用make check运行所有检查和测试。请在TravisCI之前做。
• 我们使用BestDoctor Python风格指南。
• 我们尊重Django CoC。做好事，不要做垃圾。