Skip to content

Instantly share code, notes, and snippets.

@jonahfang
Created November 23, 2022 18:53
Show Gist options
  • Save jonahfang/07f9a3cbeaac7e2ebafc3d1dec3ad0ac to your computer and use it in GitHub Desktop.
Save jonahfang/07f9a3cbeaac7e2ebafc3d1dec3ad0ac to your computer and use it in GitHub Desktop.
Django syncdb command came back for v4.1.3 version.
from django.apps import apps
from django.conf import settings
from django.core.management.base import BaseCommand
from ._create_sql_of_model import get_create_sql_for_model
from ._helper import select_by_raw_sql,exec_by_raw_sql
def _run():
for app in settings.INSTALLED_APPS:
app_name = app.split('.')[0]
app_models = apps.get_app_config(app_name).get_models()
for model in app_models:
table_name,sqls = get_create_sql_for_model(model)
if settings.DEBUG:
s = "SELECT COUNT(*) AS c FROM sqlite_master WHERE name = '%s'" % table_name
else:
s = "SELECT COUNT(*) AS c FROM information_schema.TABLES WHERE table_name='%s'" % table_name
rs = select_by_raw_sql(s)
if not rs[0]['c']:
for sql in sqls:
exec_by_raw_sql(sql)
print('CREATE TABLE DONE:%s' % table_name)
class Command(BaseCommand):
def handle(self, *args, **options):
_run()
#EOP
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)
table_name = model_state.options['db_table']
# Create a fake migration with the CreateModel operation
cm = operations.CreateModel(name=model_state.name, fields=model_state.fields.items())
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)
sqls = schema_editor.collected_sql
items = []
for sql in sqls:
if sql.startswith('--'):
continue
items.append(sql)
return table_name,items
#EOP
from django.db import connection
def select_by_raw_sql(s):
with connection.cursor() as cursor:
cursor.execute(s)
return _dictfetchall(cursor)
def exec_by_raw_sql(s):
with connection.cursor() as cursor:
cursor.execute(s)
#Thanks: https://docs.djangoproject.com/en/4.1/topics/db/sql/
def _dictfetchall(cursor):
"Return all rows from a cursor as a dict"
desc = cursor.description
return [
dict(zip([col[0] for col in desc], row))
for row in cursor.fetchall()
]
#EOP
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment