如何在第一次运行时在 Django 中创建一个 mysql 数据库？

Question

我希望我的应用程序是“即插即用”的，所以我需要在第一次运行时自动创建数据库。我使用 docker 和 docker-compose

我的尝试是在不指定数据库名称的情况下进行连接，并在运行服务器之前运行自定义命令：

command:
  sh -c "python manage.py create_db &&
         python manage.py runserver 0.0.0.0:8000"

而命令本身：

class Command(BaseCommand):
"""Django command to create DB"""

def handle(self, *args, **options):
    con = connections['default']
    db_name = os.environ.get('DB_NAME')
    db_up = False
    while not db_up:
        try:
            cursor = con.cursor()
            cursor.execute(f'CREATE DATABASE IF NOT EXISTS {db_name}')
            cursor.execute(f'USE {db_name}')
            db_up = True
        except Exception as err:
            self.stdout.write('Database unavailable, waiting 1 second...')
            self.stdout.write(str(err))
            time.sleep(1)
    self.stdout.write(self.style.SUCCESS('Database available!'))

如果这是正确的方法，那么现在我只需要更新连接以使用新创建的数据库，但我不知道如何。 cursor.execute(f'USE {db_name}') 行当然不起作用。

它是创建数据库的正确方法吗？ 如果是这样，如何更新连接？ 如果没有，怎么办？

谢谢！

编辑

在 Nealium 的提示之后，我创建了一个独立的脚本（不是 Django 命令），我在运行服务器之前运行它。

import os
import time
from MySQLdb import _mysql
import os


db_host=os.environ.get('DB_HOST')
db_user=os.environ.get('DB_USER')
db_password=os.environ.get('DB_PASS')
db_name = os.environ.get('DB_NAME')
db_up = False
while not db_up:
    try:
        db = _mysql.connect(
            host=db_host,
            user=db_user,
            password=db_password
        )
        db_up = True
        db.query(f'CREATE DATABASE IF NOT EXISTS {db_name}')
        db.close()
    except Exception as err:
        print('Database unavailable, waiting 1 second...')
        time.sleep(1)
print('Database available!')

Answer 1

这就是我的管理命令通常的样子

call_command() 基本上是 python manage.py {command}

更新了 dothing 命令

from django.core.management.base import BaseCommand
from django.core.management import call_command

def create_db():
  import mysql.connector
  try:
    mydb = mysql.connector.connect(
      host=os.environ.get('DB_HOST'),
      user=os.environ.get('DB_USER'),
      password=os.environ.get('DB_PASS')
    )

    mycursor = mydb.cursor()
    mycursor.execute('CREATE DATABASE IF NOT EXISTS {0}'.format(os.environ.get('DB_NAME')))
    return True
  except mysql.connector.Error as err:
    print('Something went wrong: {}'.format(err))
  except Exception as ex:
    message("Exception: {}".format(ex))
  return False

class Command(BaseCommand):
    help = 'does thing'

    def add_arguments(self, parser):
        # Named (optional) arguments
        parser.add_argument(
            '--import',
            action='store_true',
            help='Skips Import',
        )

    def handle(self, *args, **kwargs):
        print("Doing Thing")

        # connect to db + create if it doesn't exist
        status = create_db()

        if status:
          # create migrations
          call_command('makemigrations') # (Django Command)

          # can also pass arguemnts like a specific app
          # call_command('makemigrations', 'app1')

          # This create db **if** it doesn't exist
          #   + checks that migrations are up to date
          call_command('migrate') # (Django Command)

          if kwargs['import']:
              # another management command to handle importing
              #   I've just a csv reader and a loop
              call_command('importDb') # (Custom Command)

          # Collect Static (+ don't ask for confirmation)
          call_command('collectstatic', interactive=False) # (Django Command)

          print('Thing has been Done')
        else:
          print('Thing not Done')

因此，我只是运行：
python manage.py dothing (python manage.py dothing --import 如果我想导入 db)
接着：
python manage.py runserver 一切顺利！

---

编辑

只需执行以下操作并从设置中提取选项： https://www.w3schools.com/python/python_mysql_create_db.asp

---

编辑 2

它应该工作；根据我的测试，Django 在被告知运行查询（过滤器/获取/创建/删除）之前实际上并没有连接到数据库

您通常可以使用基本管理命令和错误的数据库设置来测试它：

1. 将设置中的数据库名称设置为“invalid_db”
2. 下面的测试命令

   from django.core.management.base import BaseCommand
   from django.core.management import call_command
   import os
   class Command(BaseCommand):
       help = 'testing '
   
       def handle(self, *args, **kwargs):
           print('Command is Running!')
           print('doing the things')
   
           print('about to migrate / use db / crash')
           print('*'*100)
           call_command('migrate')
   

   所以理论上，只要你不使用 Django 命令，你应该可以做任何你想做的事情