摘要.
# 一、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_dir 是 app 的 management目录的绝对路径，然后搜索其中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的
startapp和migrate命令，使其分别实现创建指定位置、指定模版的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
- 创建模版文件
- 文件内容
# 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的代码差异过大，有时间可以试试修改
该方法提供者
- 原文：https://juejin.im/post/6844903768270569479
- Github：https://github.com/elfgzp/django_experience.git
# 修改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