Django 模型继承并避免使用多个 django.setup() 调用

Question

更新 #1

添加了存储库结构

---

我们正在使用框架的 Django 的 ORM 部分。

我们有 1 个共享应用程序和几个应用程序，它们之间没有任何关系。共享应用程序定义了一个抽象 BaseModel，所有其他应用程序的所有其他模型都继承自该模型，以及一些共享模型，这些模型可以在其他应用程序中作为外键引用。

我们使用的是 monorepo，所以没有不同的存储库或类似的东西。

这些模型可用于各种应用程序，从（非 django）网络应用程序到处理单元，这些单元在需要时部署并在处理完成后被销毁。

我想知道是否有一个地方可以打电话

import django
django.setup()

并消除了在导入其中一个模型的每个应用程序顶部调用它的需要，这将符合我们的单元测试和服务运行。

这是共享的 models.py 文件

import json
import uuid
from datetime import datetime

import common.db.enums as shared_enums
from django.db import models
from django.db.models import CASCADE

MAX_VARCHAR_LENGTH = 255


def generate_uuid():
    return uuid.uuid4().hex


class UUIDString(models.UUIDField):  # pragma: no cover
    def from_db_value(self, value, expression, connection):
        if isinstance(value, uuid.UUID):
            return value.hex
        else:
            return value



class BaseModel(models.Model):  # pragma: no cover
    id = UUIDString(primary_key=True, default=generate_uuid, editable=False, unique=True, db_index=True)
    created_on = models.DateTimeField(default=datetime.now)
    updated_on = models.DateTimeField(auto_now_add=True)

    class Meta:
        abstract = True


class Customer(BaseModel):  # pragma: no cover
    customer_guid = UUIDString(max_length=MAX_VARCHAR_LENGTH, db_index=True, unique=True)


class FlowInstance(BaseModel):  # pragma: no cover
    name = models.CharField(max_length=MAX_VARCHAR_LENGTH, null=False)
    customer = models.ForeignKey(Customer, on_delete=CASCADE, db_index=True)
    flow_type = models.IntegerField(null=True, choices=shared_enums.FlowType.get_choices())
    created_by = models.CharField(max_length=MAX_VARCHAR_LENGTH, null=True)


class FlowRun(BaseModel):  # pragma: no cover
    flow_instance = models.ForeignKey(FlowInstance, on_delete=CASCADE, db_index=True)
    status = models.IntegerField(null=True, choices=shared_enums.ExecutionStatus.get_choices())
    start_time = models.DateTimeField(null=True)
    finish_time = models.DateTimeField(null=True)


class FlowRunResults(BaseModel):  # pragma: no cover
    flow_run = models.OneToOneField(FlowRun, on_delete=CASCADE, db_index=True)
    errors = models.TextField(null=True)
    output_location = models.TextField(null=True)

这是使用共享 models.py 的应用之一

from common.db.shared.models import BaseModel, FlowInstance, FlowRun, UUIDString, MAX_VARCHAR_LENGTH
from django.db import models
from django.db.models.deletion import CASCADE, DO_NOTHING


class Action(BaseModel):  # pragma: no cover
    action_id = models.CharField(max_length=MAX_VARCHAR_LENGTH, null=True, db_index=True)
    action_name = models.CharField(max_length=MAX_VARCHAR_LENGTH, null=True)


class DataSetDetails(BaseModel):  # pragma: no cover
    action = models.OneToOneField(Action, on_delete=DO_NOTHING, db_index=True, null=False)
    frequency = models.IntegerField(null=True)
    identified_frequency = models.IntegerField(null=True)


class RunDataSet(BaseModel):  # pragma: no cover
    dataset_details = models.ForeignKey(DataSetDetails, null=False, on_delete=CASCADE)
    flow_run = models.ForeignKey(FlowRun, on_delete=DO_NOTHING, null=True)
    storage_path = models.TextField(null=True)


class DataCollection(BaseModel):  # pragma: no cover

    name = models.CharField(max_length=MAX_VARCHAR_LENGTH, null=False)
    target = models.OneToOneField(DataSetDetails, on_delete=DO_NOTHING, null=False, related_name="target")
    related = models.OneToOneField(DataSetDetails, on_delete=DO_NOTHING, null=True, related_name="related")
    metadata = models.OneToOneField(DataSetDetails, on_delete=DO_NOTHING, null=True, related_name="metadata")
    flow_instance = models.OneToOneField(FlowInstance, on_delete=DO_NOTHING, null=False)

我们的仓库结构如下：

├── common/ <-- Modules here should be self sustainable, but can have dependencies on other commons
    ├── aws/
    ├── db/
        └── shared
            ├── migrations/
            ├── __init__.py
            ├── manage.py
            ├── models.py
            └── apps.py        
├── theme1/ <-- contain many apps related to theme1.                           
        ├── common/
            ├── theme1db/
                ├── migrations/
                ├── __init__.py
                ├── manage.py
                ├── models.py
                └── apps.py   
        ├── app1/
        └── app2/
├── theme2/ <-- contain many apps related to theme2
    ├── common/
        └── theme2db/
            ├── migrations/
            ├── __init__.py
            ├── manage.py
            ├── models.py
            └── apps.py  
        ├── service1/
        └── service2/

结构是每个目录基本上都是它自己的存储库。 每个不是根/common的目录，只能引用来自根/common的模块，或者来自主题内部的common模块

例如app1，只能引用来自theme1/app1、/theme1/common或/common的模块

Answer 1

我能想到的一个解决方案是有一个简单的 Python 文件，例如：

import django

django.setup()

然后你可以从你的其他 Python 模块中导入这个模块。由于模块无论导入多少次都只会初始化一次，django.setup() 只会被调用一次，但会在您的应用程序中运行。您可以在Python docs 中阅读更多相关信息。

另一个想法是拥有一个运行django.setup() 并充当应用程序其余部分的入口点的 Python 模块。

对于不在 Django 中的应用程序，您可能可以编写 management commands 并从这些应用程序中调用它们，但从您的问题中不清楚这是否可行。

Answer 2

您可以通过检查数据库来动态生成模型。我正在使用这种方法使用 sqlalchemy 从数据库生成模型。 This is with respect to sqlalchemy for generating models dynamically。关于 Django，我认为 this can be useful.