【问题标题】:How to programmatically generate the CREATE TABLE SQL statement for a given model in Django?如何以编程方式为 Django 中的给定模型生成 CREATE TABLE SQL 语句?
【发布时间】: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。对文档进行更多搜索导致我使用我指出的方法。

标签: python sql django migration


【解决方案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(defaultexternal_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。

    只是:

    1. 将代码保存到get_sql_create_table.py(示例)
    2. $ export DJANGO_SETTINGS_MODULE=yourproject.settings
    3. 使用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)
    

    【讨论】:

      猜你喜欢
      • 2010-09-06
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 2017-04-01
      • 2013-09-26
      • 1970-01-01
      • 1970-01-01
      相关资源
      最近更新 更多