Hướng dẫn simple workflow engine python - Python công cụ quy trình làm việc đơn giản

Daniel Gerlag - Công việc ban đầu

Giấy phép

• Dự án này được cấp phép theo giấy phép MIT - xem tệp giấy phép.md để biết chi tiết
• Liteflow
  • LiteFlow là một thư viện Python để chạy quy trình công việc. Hãy suy nghĩ: Các quy trình chạy dài với nhiều nhiệm vụ cần theo dõi trạng thái. Nó hỗ trợ các nhà cung cấp sự kiên trì và đồng thời có thể cắm được để cho phép các cụm đa nút.
  • Cài đặt
  • Mẫu
• Chào thế giới
  • Chào thế giới
  • Truyền dữ liệu
  • Truyền dữ liệu
• Sự kiện
• Sự kiên trì
• Các cụm đa nút
• Người quản lý khóa phân phối

Dự án này được cấp phép theo giấy phép MIT - xem tệp giấy phép.md để biết chi tiết

Liteflow

> pip install liteflow.core

Liteflow

LiteFlow là một thư viện Python để chạy quy trình công việc. Hãy suy nghĩ: Các quy trình chạy dài với nhiều nhiệm vụ cần theo dõi trạng thái. Nó hỗ trợ các nhà cung cấp sự kiên trì và đồng thời có thể cắm được để cho phép các cụm đa nút.

Cài đặt

Các khái niệm cơ bản

from liteflow.core import *


class Hello(StepBody):
    def run(self, context: StepExecutionContext) -> ExecutionResult:
        print("Hello world")
        
        
class Goodbye(StepBody):
    def run(self, context: StepExecutionContext) -> ExecutionResult:
        print("Goodbye")
        return ExecutionResult.next()

Các bước

class MyWorkflow(Workflow):

    def id(self):
        return "MyWorkflow"

    def version(self):
        return 1

    def build(self, builder: WorkflowBuilder):
        builder\
            .start_with(Hello)\
            .then(Goodbye)

Truyền dữ liệu giữa các bước

Kiểm soát lưu lượng

Trong khi điều kiện

Cài đặt

Các khái niệm cơ bản

Các bước

#Our workflow step with inputs and outputs
class AddNumbers(StepBody):
    def __init__(self):
        self.input1 = 0
        self.input2 = 0
        self.output = 0

    def run(self, context: StepExecutionContext) -> ExecutionResult:
        self.output = self.input1 + self.input2
        return ExecutionResult.next()


#A class to hold workflow wide data
class MyData:
    def __init__(self):
        self.value1 = 0
        self.value2 = 0
        self.value3 = 0


#Our workflow definition with mapped inputs & outputs
class MyWorkflow(Workflow):
    def build(self, builder: WorkflowBuilder):
        builder\
            .start_with(Hello)\
            .then(AddNumbers) \
                .input('input1', lambda data, context: data.value1) \
                .input('input2', lambda data, context: data.value2) \
                .output('value3', lambda step: step.output) \
            .then(Goodbye)

Mẫu

Chào thế giới

class MyWorkflow(Workflow):
    def build(self, builder: WorkflowBuilder):
        builder\
            .start_with(Hello) \
            .wait_for('event1', lambda data, context: 'key1') \
            .then(Goodbye)


#External events are published via the host
#All workflows that have subscribed to event1, key1, will be passed "hello"
host.publish_event('event1', 'key1', 'hello')

#Data from the published event can be captured and mapped to the workflow data object with an output on the WaitFor step
class MyWorkflow(Workflow):
    def build(self, builder: WorkflowBuilder):
        builder\
            .start_with(Hello) \
            .wait_for('event1', lambda data, context: 'key1') \
                .output('captured_value', lambda step: step.event_data) \
            .then(Goodbye)

Truyền dữ liệu

Chào thế giới

class DoStuff(StepBody):
    def run(self, context: StepExecutionContext) -> ExecutionResult:
        print(f"doing stuff...{context.execution_pointer.context_item}")
        return ExecutionResult.next()


class MyWorkflow(Workflow):

    def build(self, builder: WorkflowBuilder):
        builder\
            .start_with(Hello)\
            .for_each(lambda data, context: ["abc", "def", "xyz"])\
                .do(lambda x:\
                    x.start_with(DoStuff))\
            .then(Goodbye)

Truyền dữ liệu

class MyWorkflow(Workflow):
    def build(self, builder: WorkflowBuilder):
        builder\
            .start_with(Hello)\
            .while_(lambda data, context: data.value1 < 3)\
                .do(lambda do:\
                    do.start_with(DoStuff)\
                        .input('my_value', lambda data, context: data.value1)\
                        .output('value1', lambda step: step.your_value))\
            .then(Goodbye)

Truyền dữ liệu

class MyWorkflow(Workflow):

    def build(self, builder: WorkflowBuilder):
        builder\
            .start_with(Hello)\
            .if_(lambda data, context: data.value1 > 3)\
                .do(lambda x:\
                    x.start_with(DoStuff))\
            .then(Goodbye)

Sự kiện

Máy chủ lưu trữ công việc là dịch vụ chịu trách nhiệm thực hiện quy trình công việc. Nó thực hiện điều này bằng cách bỏ phiếu cho nhà cung cấp kiên trì cho các trường hợp quy trình làm việc đã sẵn sàng để chạy, thực hiện chúng và sau đó chuyển chúng trở lại cho nhà cung cấp kiên trì để được lưu trữ cho lần tiếp theo chúng được chạy. Nó cũng chịu trách nhiệm xuất bản các sự kiện cho bất kỳ quy trình công việc nào có thể đang chờ một.

Cách sử dụng

Khi ứng dụng của bạn bắt đầu, hãy tạo dịch vụ WorkflowHost bằng cách sử dụng

from liteflow.core import *


class Hello(StepBody):
    def run(self, context: StepExecutionContext) -> ExecutionResult:
        print("Hello world")
        
        
class Goodbye(StepBody):
    def run(self, context: StepExecutionContext) -> ExecutionResult:
        print("Goodbye")
        return ExecutionResult.next()

from liteflow.core import *


host = configure_workflow_host()
host.register_workflow(MyWorkflow())
host.start()

wid = host.start_workflow("MyWorkflow", 1, None)

Sự kiên trì

Vì quy trình công việc thường là các quy trình chạy dài, chúng sẽ cần phải được lưu trữ giữa các bước. Có một số nhà cung cấp kiên trì có sẵn dưới dạng các gói riêng biệt.

• Nhà cung cấp sự kiên trì bộ nhớ (nhà cung cấp mặc định, cho mục đích thử nghiệm và thử nghiệm)
• MongoDB
• (nhiều hơn nữa sẽ đến sớm ...)

Các cụm đa nút

Theo mặc định, dịch vụ WorkflowHost sẽ chạy như một nút duy nhất bằng cách sử dụng hàng đợi tích hợp và các nhà cung cấp khóa cho một cấu hình nút duy nhất. Nếu bạn muốn chạy một cụm nhiều nút, bạn sẽ cần định cấu hình một cơ chế xếp hàng bên ngoài và trình quản lý khóa phân tán để phối hợp cụm. Đây là những nhà cung cấp hiện đang có sẵn.

Nhà cung cấp hàng đợi

• SingLenodequeueProvider (Nhà cung cấp tích hợp mặc định)
• Azure
• RabbitMQ (Sắp ra mắt ...)

Người quản lý khóa phân phối

• LocallockProvider (Nhà cung cấp tích hợp mặc định)
• Azure
• RabbitMQ (Sắp ra mắt ...)

Người quản lý khóa phân phối

• LocallockProvider (Nhà cung cấp tích hợp mặc định)

• Redis Redlock (sắp ra mắt ...)

• Mẫu

• Chào thế giới

• Truyền dữ liệu

• Sự kiện

Song song foreach

• Nếu điều kiện - Initial work

Trong khi lặp lại

Tác giả