如何以编程方式为 Django 中的给定模型生成 CREATE TABLE SQL 语句？

Posted 2023-02-24

【中文标题】如何以编程方式为 Django 中的给定模型生成 CREATE TABLE SQL 语句？

【英文标题】：How to programmatically generate the CREATE TABLE SQL statement for a given model in Django?

【发布时间】：2018-07-17 21:25:11

【问题描述】：

我需要在我的 Django 应用程序中以编程方式为给定的 unmanaged 模型生成 CREATE TABLE 语句 (managed = False)

由于我正在处理旧数据库，我不想创建迁移并使用sqlmigrate。

./manage.py sql 命令可用于此目的，但已在 Django 1.8 中删除

你知道任何替代方案吗？

【问题讨论】：

什么风味数据库？有些像 mysql 允许你执行“SHOW CREATE TABLE”语句

请问，当它是非托管模型时，为什么需要“以编程方式生成”此 SQL 语句？如果重点是创建一个开发数据库，​​请从原始数据库创建一个仅模式 SQL 转储（或要求 dba 这样做）并将其包含在您的源代码中。

如果您的最终目标是创建一个表，例如对于某些测试，请查看一种强制 Django 在特定时刻为您创建它的方法，例如测试类设置，无需自己处理原始 SQL BaseDatabaseSchemaEditor.create_model(model).

@Nikita 你真的应该发布一个答案，这样任何人都可以投票给你这个好答案。如果它有效（我没有测试过），与我一年前给出的解决方案相比，这是一个很好的解决方案;-)

@fero，谢谢 :) 会的。有用。当我需要模拟外部系统的只读视图来测试我的模型时，我偶然发现了这个问题。看了你的回答，我明白了，我忘了考虑使用 Django 迁移框架 API。对文档进行更多搜索导致我使用我指出的方法。

【参考方案1】：

按照建议，我发布了该案例的完整答案，该问题可能暗示。

假设您有一个外部数据库表，您决定将其作为 Django 模型进行访问，因此将其描述为非托管模型 (Meta: managed = False)。 稍后您需要能够在代码中创建它，例如使用本地数据库进行一些测试。显然，Django 不会对非托管模型进行迁移，因此不会在您的测试数据库中创建它。 这可以使用 Django API 解决，而无需使用原始 SQL - SchemaEditor。请参阅下面更完整的示例，但作为一个简短的答案，您可以像这样使用它：

   from django.db import connections

   with connections['db_to_create_a_table_in'].schema_editor() as schema_editor:
        schema_editor.create_model(YourUnmanagedModelClass)

一个实际的例子：

# your_app/models/your_model.py

from django.db import models

class IntegrationView(models.Model):
    """A read-only model to access a view in some external DB."""

    class Meta:
        managed = False
        db_table = 'integration_view'

    name = models.CharField(
        db_column='object_name',
        max_length=255,
        primaty_key=True,
        verbose_name='Object Name',
    )
    some_value = models.CharField(
        db_column='some_object_value',
        max_length=255,
        blank=True,
        null=True,
        verbose_name='Some Object Value',
    )

    # Depending on the situation it might be a good idea to redefine
    # some methods as a NOOP as a safety-net.
    # Note, that it's not completely safe this way, but might help with some
    # silly mistakes in user code

    def save(self, *args, **kwargs):
        """Preventing data modification."""
        pass

    def delete(self, *args, **kwargs):
        """Preventing data deletion."""
        pass

现在，假设您需要能够通过 Django 创建此模型，例如进行一些测试。

# your_app/tests/some_test.py

# This will allow to access the `SchemaEditor` for the DB
from django.db import connections
from django.test import TestCase
from your_app.models.your_model import IntegrationView

class SomeLogicTestCase(TestCase):
    """Tests some logic, that uses `IntegrationView`."""

    # Since it is assumed, that the `IntegrationView` is read-only for the
    # the case being described it's a good idea to put setup logic in class 
    # setup fixture, that will run only once for the whole test case
    @classmethod
    def setUpClass(cls):
        """Prepares `IntegrationView` mock data for the test case."""

        # This is the actual part, that will create the table in the DB
        # for the unmanaged model (Any model in fact, but managed models will
        # have their tables created already by the Django testing framework)
        # Note: Here we're able to choose which DB, defined in your settings,
        # will be used to create the table

        with connections['external_db'].schema_editor() as schema_editor:
            schema_editor.create_model(IntegrationView)

        # That's all you need, after the execution of this statements
        # a DB table for `IntegrationView` will be created in the DB
        # defined as `external_db`.

        # Now suppose we need to add some mock data...
        # Again, if we consider the table to be read-only, the data can be 
        # defined here, otherwise it's better to do it in `setUp()` method.

        # Remember `IntegrationView.save()` is overridden as a NOOP, so simple
        # calls to `IntegrationView.save()` or `IntegrationView.objects.create()`
        # won't do anything, so we need to "Improvise. Adapt. Overcome."

        # One way is to use the `save()` method of the base class,
        # but provide the instance of our class
        integration_view = IntegrationView(
            name='Biggus Dickus',
            some_value='Something really important.',
        )
        super(IntegrationView, integration_view).save(using='external_db')

        # Another one is to use the `bulk_create()`, which doesn't use
        # `save()` internally, and in fact is a better solution
        # if we're creating many records

        IntegrationView.objects.using('external_db').bulk_create([
            IntegrationView(
                name='Sillius Soddus',
                some_value='Something important',
            ),
            IntegrationView(
                name='Naughtius Maximus',
                some_value='Whatever',
            ),
        ])

    # Don't forget to clean after
    @classmethod
    def tearDownClass(cls):
        with connections['external_db'].schema_editor() as schema_editor:
            schema_editor.delete_model(IntegrationView)

    def test_some_logic_using_data_from_integration_view(self):
         self.assertTrue(IntegrationView.objects.using('external_db').filter(
             name='Biggus Dickus',
         ))

为了使示例更完整...由于我们使用多个 DB（default 和 external_db），Django 将尝试在这两个数据库上运行迁移以进行测试，到目前为止，DB 中没有选项设置以防止这种情况。所以我们必须使用自定义的 DB 路由器进行测试。

 # your_app/tests/base.py

class PreventMigrationsDBRouter:
    """DB router to prevent migrations for specific DBs during tests."""
    _NO_MIGRATION_DBS = 'external_db', 

    def allow_migrate(self, db, app_label, model_name=None, **hints):
        """Actually disallows migrations for specific DBs."""
        return db not in self._NO_MIGRATION_DBS

以及所描述案例的测试设置文件示例：

# settings/test.py

DATABASES = 
    'default': 
        'ENGINE': 'django.db.backends.oracle',
        'NAME': 'db_name',
        'USER': 'username',
        'HOST': 'localhost',
        'PASSWORD': 'password',
        'PORT': '1521',
    ,
    # For production here we would have settings to connect to the external DB,
    # but for testing purposes we could get by with an SQLite DB 
    'external_db': 
        'ENGINE': 'django.db.backends.sqlite3',
    ,


# Not necessary to use a router in production config, since if the DB 
# is unspecified explicitly for some action Django will use the `default` DB
DATABASE_ROUTERS = ['your_app.tests.base.PreventMigrationsDBRouter', ]

希望这个详细的新 Django 用户友好示例能够帮助某人并节省他们的时间。

【参考方案2】：

不幸的是，似乎没有简单的方法可以做到这一点，但为了你的运气，我刚刚成功地为你在 django 迁移丛林的内部挖掘工作的 sn-p。

只是：

将代码保存到get_sql_create_table.py（示例） 做$ export DJANGO_SETTINGS_MODULE=yourproject.settings 使用python get_sql_create_table.py yourapp.yourmodel 启动脚本

它应该输出你需要的东西。

希望对你有帮助！

import django
django.setup()

from django.db.migrations.state import ModelState
from django.db.migrations import operations
from django.db.migrations.migration import Migration
from django.db import connections
from django.db.migrations.state import ProjectState

def get_create_sql_for_model(model):

    model_state = ModelState.from_model(model)

    # Create a fake migration with the CreateModel operation
    cm = operations.CreateModel(name=model_state.name, fields=model_state.fields)
    migration = Migration("fake_migration", "app")
    migration.operations.append(cm)

    # Let the migration framework think that the project is in an initial state
    state = ProjectState()

    # Get the SQL through the schema_editor bound to the connection
    connection = connections['default']
    with connection.schema_editor(collect_sql=True, atomic=migration.atomic) as schema_editor:
        state = migration.apply(state, schema_editor, collect_sql=True)

    # return the CREATE TABLE statement
    return "\n".join(schema_editor.collected_sql)

if __name__ == "__main__":

    import importlib
    import sys

    if len(sys.argv) < 2:
        print("Usage:  <app.model>".format(sys.argv[0]))
        sys.exit(100)

    app, model_name = sys.argv[1].split('.')

    models = importlib.import_module(".models".format(app))
    model = getattr(models, model_name)
    rv = get_create_sql_for_model(model)
    print(rv)