Dagster与Django的集成

Question

嗨，我正在尝试将达格斯特集成到正在进行的Django项目中。我有点挣扎于提供Django上下文(模型，应用程序，.)敬达格斯特。到目前为止，我只是在检查dagit是否存在于sys.argv[0] in init__.py中，这些应用程序都在使用Dagster。

<!-- language: python -->
## project/app/__init__.py
import sys

import django

if 'dagit-cli' in sys.argv[0]:
    print('dagit')
    django.setup()

有人能帮我安排一下吗？

Answer 1

我还会像Timothé说的那样使用Django自定义管理命令在Django的上下文中执行函数或访问ORM。

但是，如果需要用例直接访问Django ORM，则需要执行几个步骤：

1. 将Django项目添加到sys.path中，以便Python可以找到包。
2. 设置Django环境。

import os
import sys

import django
from dagster import execute_pipeline, pipeline, solid


# Add the project to sys.path, so that Python can find packages
PROJECT_ROOT = os.path.join(os.path.dirname(__file__), 'demo')
sys.path.append(PROJECT_ROOT)

# Set up the Django environment
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'demo.settings')
django.setup()


from customers.models import Customer


@solid
def hello_django_orm(context):
    number_of_customers = Customer.objects.count()
    message = f'Found {number_of_customers} customers!'
    context.log.info(message)
    return number_of_customers


@pipeline
def hello_django_orm_pipeline():
    hello_django_orm()


if __name__ == '__main__':
    result = execute_pipeline(hello_django_orm_pipeline)
    assert result.success

见下面的结果。

​

​

请参阅完整的示例这里。

Answer 2

据我所知，没有高级集成，但我找到了从django视图启动管道的一些解决办法：

• 通过从python代码中调用包函数execute_pipeline(<your_pipeline>)来同步运行管道，例如在View内部(即使它在同步/异步范例方面是非意义的)。
• 创建一个自定义管理命令，并使用django.core.management.call_command()钩子从python代码中调用它