python

Django 源码学习

文章暂存

systemime
2020-09-24
25 min

摘要.

# 一、manage.py学习

  • Django 3.1.1

# 1. 入口


django所有的命令操作都是入口都是 manage.py 文件,启动一个django项目,makemigrations migrate startapp,这三个命令几乎贯穿整个开发过程,从一次改造migrate 命令的想法开始,研究一下django对命令对处理过程

# 2. manage.py


如下是django 3.1.1的manage.py文件内容,main()中,首先定义了django项目的用户配置文件,通过execute_from_command_line方法将命令行作为参数传递

#!/usr/bin/env python
"""Django's command-line utility for administrative tasks."""
import os
import sys


def main():
    """Run administrative tasks."""
    os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'day01.settings.base')
    try:
        from django.core.management import execute_from_command_line
    except ImportError as exc:
        raise ImportError(
            "Couldn't import Django. Are you sure it's installed and "
            "available on your PYTHONPATH environment variable? Did you "
            "forget to activate a virtual environment?"
        ) from exc
    execute_from_command_line(sys.argv)


if __name__ == '__main__':
    main()

# 3. execute_from_command_line

  • django/core/management/__init__.py


里面调用了 ManagementUtility 类中的 execute 方法

def execute_from_command_line(argv=None):
    """Run a ManagementUtility."""
    utility = ManagementUtility(argv)
    utility.execute()


execute 中主要是解析了传入的参数 sys.argv ,最终调用了get_command()处理参数

# 4. execute执行过程

  • django/core/management/__init__.py

<br />

    def execute(self):
        """
        给定命令行参数,找出命令,给这个命令创建一个解析器并运行它.
        """
        try:
            subcommand = self.argv[1]
        except IndexError:
            subcommand = 'help'  # 没有传入命令显示help内容

        # 预处理提取 --settings 和 --pythonpath 这俩项配置.
        # 这些参数会影响命令的可用性,因此它们必须要提前处理,当然了你不传也没任何影响.
        parser = CommandParser(usage='%(prog)s subcommand [options] [args]', add_help=False, allow_abbrev=False)
        parser.add_argument('--settings')
        parser.add_argument('--pythonpath')
        parser.add_argument('args', nargs='*')  # catch-all
        try:  # 处理已知参数
            # 返回命令命名空间和参数
            options, args = parser.parse_known_args(self.argv[2:])
            # 在此处包括所有命令应接受的所有默认选项
            # 以便ManagementUtility可以在搜索用户命令之前对其进行处理。
            handle_default_options(options)
        except CommandError:
            pass  # Ignore any option errors at this point.

        try:  # 检查settings能否访问定义的app
            settings.INSTALLED_APPS
        except ImproperlyConfigured as exc:
            self.settings_exception = exc
        except ImportError as exc:
            self.settings_exception = exc

        if settings.configured:  # 如果配置了settings则返回True,此处逻辑稍微复杂,以后再看
            # 即使代码已损坏,也要启动自动重新加载的dev服务器.
            # The hardcoded condition is a code smell but we can't rely on a
            # flag on the command class because we haven't located it yet.
            if subcommand == 'runserver' and '--noreload' not in self.argv:
                # 如果是运行命令(默认wsgi)
                try:
                    # 自动检查错误并装载
                    autoreload.check_errors(django.setup)()
                except Exception:  # 启动失败的异常处理
                    # The exception will be raised later in the child process
                    # started by the autoreloader. Pretend it didn't happen by
                    # loading an empty list of applications.
                    apps.all_models = defaultdict(dict)
                    apps.app_configs = {}
                    apps.apps_ready = apps.models_ready = apps.ready = True

                    # Remove options not compatible with the built-in runserver
                    # (e.g. options for the contrib.staticfiles' runserver).
                    # Changes here require manually testing as described in
                    # #27522.
                    _parser = self.fetch_command('runserver').create_parser('django', 'runserver')
                    _options, _args = _parser.parse_known_args(self.argv[2:])
                    for _arg in _args:
                        self.argv.remove(_arg)

            # 其他情况下django.setup必然是成功的
            else:
                django.setup()
		
        # 输出一些建议内容
        self.autocomplete()

        if subcommand == 'help':  # 如果是help命令
            if '--commands' in args: # 同时commands在参数列表中
                # 返回所有系统命令(可能是系统app?理解可能不到位)
                sys.stdout.write(self.main_help_text(commands_only=True) + '\n')
            elif not options.args:
                # 返回所有命令(分模块)
                sys.stdout.write(self.main_help_text() + '\n')
            else:
                self.fetch_command(options.args[0]).print_help(self.prog_name, options.args[0])
        # Special-cases: We want 'django-admin --version' and
        # 'django-admin --help' to work, for backwards compatibility.
        elif subcommand == 'version' or self.argv[1:] == ['--version']:
            # 输出版本,没什么好说的
            sys.stdout.write(django.get_version() + '\n')
        elif self.argv[1:] in (['--help'], ['-h']):
            # 这个地方有点迷,就是为了支持--help, -h, help?,这个地方完全可以写在上个if中
            sys.stdout.write(self.main_help_text() + '\n')
        else:
            # 如果不是以上这些命令,那么继续往下找
            self.fetch_command(subcommand).run_from_argv(self.argv)

self``.fetch_command``_(_``subcommand``_)_``.run_from_argv``_(_``self``.argv``_)_是一个链式操作,从fetch_command 开始看

    def fetch_command(self, subcommand):
        """
        尝试获取给定的子命令,如果找不到,
        则使用从命令行调用的相应命令(通常为“ django-admin”或“ manage.py”)打印一条消息
        
        通过get_command方法遍历所有注册的 INSTALLED_APPS 路径下的management 
        寻找 (find_commands) 用户自定义的命令。
    	是一个字典集合,key 为子命令名称,value 为 app 的名称
        """
        # 将字典映射命令名称返回到其回调应用程序
        # 这里建议先往下看get_commands实现
        commands = get_commands()
        try:
            # 获取到app的名称(INSTALLED_APPS定义的名称)
            app_name = commands[subcommand]
        except KeyError:
            # 如果不存在的命令,或者没有指定django的settings触发异常告知用户
            if os.environ.get('DJANGO_SETTINGS_MODULE'):
                # 如果发生异常,退出并进行提示 (并根据当前错误命令,提示与其相关的命令)
                # 这里用到名为 difflib 模块(标准库下的)
                settings.INSTALLED_APPS
            elif not settings.configured:
                sys.stderr.write("No Django settings specified.\n")
           
            possible_matches = get_close_matches(subcommand, commands)
            sys.stderr.write('Unknown command: %r' % subcommand)
            # 给出联想猜测的用户想输入的命令
            if possible_matches:
                sys.stderr.write('. Did you mean %s?' % possible_matches[0])
            sys.stderr.write("\nType '%s help' for usage.\n" % self.prog_name)
            sys.exit(1)
            
        # 调用者的基类为 BaseCommand,所有handle的子类的父类,startapp、migrate等类均继承此类
        if isinstance(app_name, BaseCommand):
            # 如果命令已加载则直接使用
            klass = app_name
        else:
            # 否则导入并返回Command实例类
            klass = load_command_class(app_name, subcommand)
        return klass

get_commands 通过 find_commands 去查询每个app下commands目录下不以"_"开头的文件名。
也是自定义命令的实现方式,实际上这些都是Command 类,其中的handler的方法会在后面被调用,最终实现该命令

@functools.lru_cache(maxsize=None)
def get_commands():
    """
    将字典映射命令名称返回到其回调应用程序

    在django.core中查找management.commands软件包,
    然后在每个已安装的应用程序中查找-如果存在命令软件包,则在该软件包中注册所有命令

    始终包含核心命令. 如果已指定设置模块,则还包括用户定义的命令.

    字典的格式为{command_name:app_name}. 
    然后,可以在调用load_command_class(app_name,command_name)的过程中使用此字典中的键/值对。

    如果必须加载命令的特定版本(例如,使用startapp命令),
    则可以将实例化的模块放置在词典中以代替应用程序名称

    该字典在第一个调用中缓存,并在后续调用中重用。
    """
    # find_commands源码在下方
    # 获取到字典
    commands = {name: 'django.core' for name in find_commands(__path__[0])}

    # 如果没有配置项目settings(即用户自定义settings)
    if not settings.configured:
        return commands

    for app_config in reversed(list(apps.get_app_configs())):
        path = os.path.join(app_config.path, 'management')
        commands.update({name: app_config.name for name in find_commands(path)})
	# 最后的这个字典,key是每个app下management.commands下命令文件名
    # 
    return commands

management_dirappmanagement目录的绝对路径,然后搜索其中commands目录不以_开头的文件命令

def find_commands(management_dir):
    """
    Given a path to a management directory, return a list of all the command
    names that are available.
    """
    command_dir = os.path.join(management_dir, 'commands')
    return [name for _, name, is_pkg in pkgutil.iter_modules([command_dir])
            if not is_pkg and not name.startswith('_')]


**fetch_command**部分结束,接下来是链式操作的**run_from_argv**阶段

    def run_from_argv(self, argv):
        """
        设置所需的任何环境变量(e.g., Python 路径 和 Django settings), 
        然后运行它.如果命令引发CommandError错误,则将其拦截并将其打印到stderr。
        如果存在--traceback选项或引发的Exception不是CommandError,则引发它。
        """
        self._called_from_command_line = True
        # 创建并返回 ArgumentParser类,它将用于解析此命令的参数。
        parser = self.create_parser(argv[0], argv[1])
		# 捕获缺少的参数,以便给出更好的错误提示
        options = parser.parse_args(argv[2:])
        # 获取options内容
        cmd_options = vars(options)
        
        # 兼容旧的optparse
        args = cmd_options.pop('args', ())
        # 此处包装所有命令应接受的所有默认选项
        # 以便ManagementUtility可以在搜索用户命令之前对其进行处理。
        handle_default_options(options)
        try:
            # 开始执行命令方法,由该execute调用BaseCommand子类对应命令的handle方法
            self.execute(*args, **cmd_options)
        except CommandError as e:
            # 就是一些异常处理了
            if options.traceback:
                raise

            # SystemCheckError takes care of its own formatting.
            if isinstance(e, SystemCheckError):
                self.stderr.write(str(e), lambda x: x)
            else:
                self.stderr.write('%s: %s' % (e.__class__.__name__, e))
            sys.exit(e.returncode)
        finally:  # 结束以后关闭数据库链接并无论是否建立链接(你们在偷懒-·|·-)
            try:
                connections.close_all()
            except ImproperlyConfigured:
                # Ignore if connections aren't setup at this point (e.g. no
                # configured settings).
                pass
    def execute(self, *args, **options):
        """
        尝试执行此命令,并在需要时执行系统检查(
        由“ requires_system_checks”属性控制,除非强制跳过)
        """
        
        # 设置颜色相关
        if options['force_color'] and options['no_color']:
            raise CommandError("The --no-color and --force-color options can't be used together.")
        if options['force_color']:
            self.style = color_style(force_color=True)
        elif options['no_color']:
            self.style = no_style()
            self.stderr.style_func = None
        if options.get('stdout'):
            self.stdout = OutputWrapper(options['stdout'])
        if options.get('stderr'):
            self.stderr = OutputWrapper(options['stderr'])

        # 是否跳过检查之类
        if self.requires_system_checks and not options['skip_checks']:
            self.check()
        if self.requires_migrations_checks:
            self.check_migrations()
        # 正式执行命令的handle方法,如果想自定义命令或者重载django已有的命令
        # 目录结构正确位置,并继承BaseCommand方法,实现或重载自己的handle方法
        output = self.handle(*args, **options)
        # 一般情况下不会返回output,如果遇到了再进行补充
        if output:
            if self.output_transaction:
                connection = connections[options.get('database', DEFAULT_DB_ALIAS)]
                output = '%s\n%s\n%s' % (
                    self.style.SQL_KEYWORD(connection.ops.start_transaction_sql()),
                    output,
                    self.style.SQL_KEYWORD(connection.ops.end_transaction_sql()),
                )
            self.stdout.write(output)
        return output

# 二、重载实现


第一部分中,从execute中执行命令类的handle方法,实现对应命令的功能。

基于这个入口,重载一下Django的startappmigrate命令,使其分别实现创建指定位置、指定模版的app,以及自动迁移所有数据库(多数据库情况下)

# 1. 准备工作

# 注册自定义命令

  • 在 settings.py 中添加
# django项目目录
BASE_DIR = Path(__file__).resolve(strict=True).parent.parent.parent

# app目录
PROJECT_DIR = Path(__file__).resolve(strict=True).parent.parent

INSTALLED_APPS = [
    'django.contrib.contenttypes',

    # 自定义命令
    # yobee_db 是项目根目录文件夹
    'yobee_db.libs.commands',
    ...
]
  • 目录结构
.
├── README.md
├── manage.py
├── requirements.txt
└── yobee_db
    ├── __init__.py
    ├── db
    │   ├── __init__.py
    │   └── test_db
    ├── libs
    │   ├── __init__.py
    │   ├── app_template
    │   ├── commands
    └── settings
        ├── __init__.py
        └── dev_settings.py
  • 创建主要路径及文件
    • Your_project_path/apps/libs/commands/management/comment/startapp.py
    • Your_project_path/apps/libs/commands/management/comment/migrate.py
  • 创建模版文件

image.png

  • 文件内容
# models.py-tpl

from django.db import models
from yobee_db.libs.db.models import BaseModel

# Create your models here.
----------------------------
# apps.py-tpl

from django.apps import AppConfig


class {{ camel_case_app_name }}Config(AppConfig):
    name = 'yobee_db.db.{{ app_name }}'
----------------------------
# __init__.py-tpl

# 为空
----------------------------
# __init__.py-tpl

# 为空

# 2. 重载startapp命令

  • 目标:执行 python manage.py startapp xxx 命令后
    • app自动生成到定义的目录位置
    • app目录结构是我们自定义的结构
# startapp.py中添加如下内容

"""覆盖django startapp 命令"""
from django.conf import settings
from django.core.management import CommandError
from django.core.management.templates import TemplateCommand


class Command(TemplateCommand):
    help = (
        "Creates a Django app directory structure for the given app name in "
        "the current directory or optionally in the given directory."
    )
    missing_args_message = "You must provide an application name."

    def handle(self, **options):
        app_name = options.pop("name")

        if options.pop("directory"):
            raise CommandError("custom directory is not allowed")
        if options.pop("template"):
            raise CommandError("custom template is not allowed")

        target = settings.PROJECT_DIR / f"db/{app_name}"  # 生成的app所在位置

        if not target.exists():
            # 参数说明: 如果中间路径不存在,则创建它
            # 同时忽略存在部分路径错误,不引发FileExistsError错误
            target.mkdir(parents=True, exist_ok=True)

        target = str(target)
        template = settings.PROJECT_DIR / "libs/app_template"  使用的app生成模版
        options['template'] = str(template)
        super().handle("app", app_name, target, **options)  # 调用父类的方法

# 3. 重载migrate命令

  • 目标:多数据库情况下,执行 **python manage.py migrate**
    • 自动迁移所有或指定的多个数据库
    • **兼容 ****--database** 等其他参数

**

# 注意 django/core/management/base.py 中执行命令的方法

output = self.handle(*args, **options)

# 以及run_from_argv方法中,执行完成后的数据库关闭操作
......
		finally:
            try:
                connections.close_all()
            except ImproperlyConfigured:
                # Ignore if connections aren't setup at this point (e.g. no
                # configured settings).
                pass
# migrate.py
# -*- coding: utf-8 -*-
"""增强django migrate 命令
自动迁移所有数据库,由routers控制迁移保证迁移到指定数据库
"""
import re

from django.conf import settings
from django.core.exceptions import ImproperlyConfigured
from django.core.management.commands.migrate import Command as MigrateCommand
from django.db import connections


class Command(MigrateCommand):
    """
    python manage.py migrate 顺序迁移所有app的数据库模型
    默认迁移顺序为settings的DATABASES配置顺序(除default外)

    兼容--database,使用指定迁移多个数据库,使用英文","分割,迁移顺序为书写顺序
    --database=admin_db,agent_db,test_db

    当migrate后附加除--database以外参数时,建议使用单数据库操作
    否则,如fake、run-syncdb等参数可能被应用到所有数据库

    django/db/utils.py已处理可能出现的错误,无需此处定义
    """

    help = "Updates database schema. " \
           "Manages both apps with migrations and those without."

    db_more = re.compile(r"[\w]+,\w+")
    db_lists = [db for db in settings.DATABASES if db != "default"]

    def handle(self, *args, **options):
        if options['database'] == "default":  # 未指定迁移的数据库,迁移所有数据库
            db_lists = self.db_lists
        elif self.db_more.match(options['database']): # 指定多个数据库但不是全部数据库
            db_lists = options["database"].split(",")
        else:
            db_lists = [options["database"]]  # 单个数据情况

        for db in db_lists:
            options['database'] = db
            super().handle(self, *args, **options)
            # 20/11/03 重载命令实际上也是run_from_argv 方法中调用的,
            # 调用结束会自动执行 下面代码,
            # connections是一个数据库链接的字典,会关闭所有链接
            # try:
            #     connections.close_all()
            # except ImproperlyConfigured:
            #     # Ignore if connections aren't setup at this point (e.g. no
            #     # configured settings).
            #     pass

这里不需要处理迁移数据库不存在或者没有配置等错误信息,因为我们在处理好最初的参数部分后,本质去循环执行原来migrate的handle方法,其中包含了可能出现的任何问题,我们唯一需要注意的是在每一次循环后,都关闭数据库链接,因为现在是多数据库的情况。

如果你把migrate.py修改为my_migrate.py,等于你不再是重写migrate命令,而是获得了一个新的自定义的my_migrate命令,注意不能以"_"开头文件,前面源码里说明了的

你也可以重写整个类,直接继承 BaseCommand 类重写整个migrate方法,这样你就可以在handle中直接重写connection 的处理方法
附migrate主要方法源码

class Command(BaseCommand):
    help = "Updates database schema. Manages both apps with migrations and those without."
    requires_system_checks = False

    def add_arguments(self, parser):
        ......
    
        @no_translations
    def handle(self, *args, **options):
        database = options['database']
        if not options['skip_checks']:
            self.check(databases=[database])

        self.verbosity = options['verbosity']
        self.interactive = options['interactive']

        # Import the 'management' module within each installed app, to register
        # dispatcher events.
        for app_config in apps.get_app_configs():
            if module_has_submodule(app_config.module, "management"):
                import_module('.management', app_config.name)

        # Get the database we're operating from
        connection = connections[database]

        # Hook for backends needing any database preparation
        connection.prepare_database()
        # Work out which apps have migrations and which do not
        executor = MigrationExecutor(connection, self.migration_progress_callback)

        # Raise an error if any migrations are applied before their dependencies.
        executor.loader.check_consistent_history(connection)

        # Before anything else, see if there's conflicting apps and drop out
        # hard if there are any
        conflicts = executor.loader.detect_conflicts()
        if conflicts:
            name_str = "; ".join(
                "%s in %s" % (", ".join(names), app)
                for app, names in conflicts.items()
            )
            raise CommandError(
                "Conflicting migrations detected; multiple leaf nodes in the "
                "migration graph: (%s).\nTo fix them run "
                "'python manage.py makemigrations --merge'" % name_str
            )

        # If they supplied command line arguments, work out what they mean.
        run_syncdb = options['run_syncdb']
        target_app_labels_only = True
        if options['app_label']:
            # Validate app_label.
            app_label = options['app_label']
            try:
                apps.get_app_config(app_label)
            except LookupError as err:
                raise CommandError(str(err))
            if run_syncdb:
                if app_label in executor.loader.migrated_apps:
                    raise CommandError("Can't use run_syncdb with app '%s' as it has migrations." % app_label)
            elif app_label not in executor.loader.migrated_apps:
                raise CommandError("App '%s' does not have migrations." % app_label)

        if options['app_label'] and options['migration_name']:
            migration_name = options['migration_name']
            if migration_name == "zero":
                targets = [(app_label, None)]
            else:
                try:
                    migration = executor.loader.get_migration_by_prefix(app_label, migration_name)
                except AmbiguityError:
                    raise CommandError(
                        "More than one migration matches '%s' in app '%s'. "
                        "Please be more specific." %
                        (migration_name, app_label)
                    )
                except KeyError:
                    raise CommandError("Cannot find a migration matching '%s' from app '%s'." % (
                        migration_name, app_label))
                targets = [(app_label, migration.name)]
            target_app_labels_only = False
        elif options['app_label']:
            targets = [key for key in executor.loader.graph.leaf_nodes() if key[0] == app_label]
        else:
            targets = executor.loader.graph.leaf_nodes()

        plan = executor.migration_plan(targets)
        exit_dry = plan and options['check_unapplied']

        if options['plan']:
            self.stdout.write('Planned operations:', self.style.MIGRATE_LABEL)
            if not plan:
                self.stdout.write('  No planned migration operations.')
            for migration, backwards in plan:
                self.stdout.write(str(migration), self.style.MIGRATE_HEADING)
                for operation in migration.operations:
                    message, is_error = self.describe_operation(operation, backwards)
                    style = self.style.WARNING if is_error else None
                    self.stdout.write('    ' + message, style)
            if exit_dry:
                sys.exit(1)
            return
        if exit_dry:
            sys.exit(1)

        # At this point, ignore run_syncdb if there aren't any apps to sync.
        run_syncdb = options['run_syncdb'] and executor.loader.unmigrated_apps
        # Print some useful info
        if self.verbosity >= 1:
            self.stdout.write(self.style.MIGRATE_HEADING("Operations to perform:"))
            if run_syncdb:
                if options['app_label']:
                    self.stdout.write(
                        self.style.MIGRATE_LABEL("  Synchronize unmigrated app: %s" % app_label)
                    )
                else:
                    self.stdout.write(
                        self.style.MIGRATE_LABEL("  Synchronize unmigrated apps: ") +
                        (", ".join(sorted(executor.loader.unmigrated_apps)))
                    )
            if target_app_labels_only:
                self.stdout.write(
                    self.style.MIGRATE_LABEL("  Apply all migrations: ") +
                    (", ".join(sorted({a for a, n in targets})) or "(none)")
                )
            else:
                if targets[0][1] is None:
                    self.stdout.write(
                        self.style.MIGRATE_LABEL('  Unapply all migrations: ') +
                        str(targets[0][0])
                    )
                else:
                    self.stdout.write(self.style.MIGRATE_LABEL(
                        "  Target specific migration: ") + "%s, from %s"
                        % (targets[0][1], targets[0][0])
                    )

        pre_migrate_state = executor._create_project_state(with_applied_migrations=True)
        pre_migrate_apps = pre_migrate_state.apps
        emit_pre_migrate_signal(
            self.verbosity, self.interactive, connection.alias, apps=pre_migrate_apps, plan=plan,
        )

        # Run the syncdb phase.
        if run_syncdb:
            if self.verbosity >= 1:
                self.stdout.write(self.style.MIGRATE_HEADING("Synchronizing apps without migrations:"))
            if options['app_label']:
                self.sync_apps(connection, [app_label])
            else:
                self.sync_apps(connection, executor.loader.unmigrated_apps)

        # Migrate!
        if self.verbosity >= 1:
            self.stdout.write(self.style.MIGRATE_HEADING("Running migrations:"))
        if not plan:
            if self.verbosity >= 1:
                self.stdout.write("  No migrations to apply.")
                # If there's changes that aren't in migrations yet, tell them how to fix it.
                autodetector = MigrationAutodetector(
                    executor.loader.project_state(),
                    ProjectState.from_apps(apps),
                )
                changes = autodetector.changes(graph=executor.loader.graph)
                if changes:
                    self.stdout.write(self.style.NOTICE(
                        "  Your models have changes that are not yet reflected "
                        "in a migration, and so won't be applied."
                    ))
                    self.stdout.write(self.style.NOTICE(
                        "  Run 'manage.py makemigrations' to make new "
                        "migrations, and then re-run 'manage.py migrate' to "
                        "apply them."
                    ))
            fake = False
            fake_initial = False
        else:
            fake = options['fake']
            fake_initial = options['fake_initial']
        post_migrate_state = executor.migrate(
            targets, plan=plan, state=pre_migrate_state.clone(), fake=fake,
            fake_initial=fake_initial,
        )
        # post_migrate signals have access to all models. Ensure that all models
        # are reloaded in case any are delayed.
        post_migrate_state.clear_delayed_apps_cache()
        post_migrate_apps = post_migrate_state.apps

        # Re-render models of real apps to include relationships now that
        # we've got a final state. This wouldn't be necessary if real apps
        # models were rendered with relationships in the first place.
        with post_migrate_apps.bulk_update():
            model_keys = []
            for model_state in post_migrate_apps.real_models:
                model_key = model_state.app_label, model_state.name_lower
                model_keys.append(model_key)
                post_migrate_apps.unregister_model(*model_key)
        post_migrate_apps.render_multiple([
            ModelState.from_model(apps.get_model(*model)) for model in model_keys
        ])

        # Send the post_migrate signal, so individual apps can do whatever they need
        # to do at this point.
        emit_post_migrate_signal(
            self.verbosity, self.interactive, connection.alias, apps=post_migrate_apps, plan=plan,
        )

        def migration_progress_callback(self, action, migration=None, fake=False):
            ......
            
        def sync_apps(self, connection, app_labels):
            ......
            
        @staticmethod
    	def describe_operation(operation, backwards):
            ......


一个django1.11版本的实现方法,但是不适用于现在的django版本,因为要改写东西太多了, 从下面第39行开始,已经和现版本django的代码差异过大,有时间可以试试修改

该方法提供者

# 修改handle方法,注意这段代码

# Get the database we're operating from
connection = connections[database]

# Hook for backends needing any database preparation
connection.prepare_database()
# Work out which apps have migrations and which do not
executor = MigrationExecutor(connection, self.migration_progress_callback)

=================开始修改=======================

	def handle(self, *args, **options):
    	......
		db_routers = [import_string(router)() for router in conf.settings.DATABASE_ROUTERS]
        for connection in connections.all():
            # Hook for backends needing any database preparation
            connection.prepare_database()
            # Work out which apps have migrations and which do not
            executor = MigrationExecutor(connection, self.migration_progress_callback)
            # Raise an error if any migrations are applied before their dependencies.
            executor.loader.check_consistent_history(connection)

            # Before anything else, see if there's conflicting apps and drop out
            # hard if there are any
            conflicts = executor.loader.detect_conflicts()
            if conflicts:
                name_str = "; ".join(
                    "%s in %s" % (", ".join(names), app)
                    for app, names in conflicts.items()
                )
                raise CommandError(
                    "Conflicting migrations detected; multiple leaf nodes in the "
                    "migration graph: (%s).\nTo fix them run "
                    "'python manage.py makemigrations --merge'" % name_str
                )

            # If they supplied command line arguments, work out what they mean.
            targets, target_app_labels_only = self._get_targets(connection, executor, db_routers, options)
            ......
     
    def _get_targets(self, connection, executor, db_routers, options):
        target_app_labels_only = True
        if options['migration_name']:
            app_label, migration_name = options['app_label'], options['migration_name']
            if app_label not in executor.loader.migrated_apps:
                raise CommandError(
                    "App '%s' does not have migrations." % app_label
                )
            if migration_name == "zero":
                targets = [(app_label, None)]
            else:
                try:
                    migration = executor.loader.get_migration_by_prefix(app_label, migration_name)
                except AmbiguityError:
                    raise CommandError(
                        "More than one migration matches '%s' in app '%s'. "
                        "Please be more specific." %
                        (migration_name, app_label)
                    )
                except KeyError:
                    raise CommandError("Cannot find a migration matching '%s' from app '%s'." % (
                        migration_name, app_label))
                targets = [(app_label, migration.name)]
            target_app_labels_only = False
        else:
            targets = executor.loader.graph.leaf_nodes()

        targets = self._filter_targets(connection, targets, db_routers)
        return targets, target_app_labels_only

# 4. 额外话题 做为其他django项目的依赖

如果你的django项目是作为其他项目的依赖或者导入,比如本例子的db、lib目录是其他django项目的ORM数据库依赖(其他django项目依赖这一个django项目的数据库,便于数据库控制)

你需要一个额外的 MANIFEST.in 文件,详见: django进阶指南

prune yobee_db/libs/app_template
prune yobee_db/libs/commands
prune yobee_db/db/*/migrations

如果你的这个项目在github/gitlab上,你可以生成这个项目的token或者打包为pip包,安装方法为

pip install git+https://deploy:You_project_token@github.com/systemime/django-db@xxxxxx#egg=django-db
上次编辑于: 2021/5/20 下午3:26:49