Управление Django-проектом через manage.py: от основ до продвинутых техник

Расширенные примеры работы с manage.py

Пример 1. Кастомная команда с обработкой ошибок и логированием

Создадим команду, которая импортирует данные из CSV в модель Product. Допустим, есть модель:

# myapp/models.py
from django.db import models

class Product(models.Model):
    name = models.CharField(max_length=100)
    price = models.DecimalField(max_digits=10, decimal_places=2)
    created_at = models.DateTimeField(auto_now_add=True)

Команда:

# myapp/management/commands/import_products.py
import csv
import logging
from django.core.management.base import BaseCommand, CommandError
from myapp.models import Product

logger = logging.getLogger(__name__)

class Command(BaseCommand):
    help = 'Импортирует товары из CSV файла'

    def add_arguments(self, parser):
        parser.add_argument('csv_file', type=str, help='Путь к CSV файлу')
        parser.add_argument('--delimiter', type=str, default=',', help='Разделитель полей (по умолчанию запятая)')
        parser.add_argument('--dry-run', action='store_true', help='Тестовый прогон без сохранения')

    def handle(self, *args, **options):
        file_path = options['csv_file']
        delimiter = options['delimiter']
        dry_run = options['dry_run']

        try:
            with open(file_path, newline='', encoding='utf-8') as f:
                reader = csv.DictReader(f, delimiter=delimiter)
                if 'name' not in reader.fieldnames or 'price' not in reader.fieldnames:
                    raise CommandError('Ожидаются столбцы "name" и "price"')

                count = 0
                for row in reader:
                    name = row['name'].strip()
                    price = row['price'].strip()
                    if not name or not price:
                        logger.warning(f'Пропущена строка {row}')
                        continue
                    try:
                        price = float(price)
                    except ValueError:
                        raise CommandError(f'Неверный формат цены: {price}')

                    if not dry_run:
                        Product.objects.get_or_create(name=name, defaults={'price': price})
                    count += 1

                self.stdout.write(self.style.SUCCESS(f'Обработано {count} записей. Dry-run: {dry_run}'))
        except FileNotFoundError:
            raise CommandError(f'Файл {file_path} не найден')

Пример вызова:

python manage.py import_products products.csv --delimiter ';' --dry-run
# Вывод (если ошибок нет):
# Обработано 10 записей. Dry-run: True

При реальном импорте (без --dry-run) записи будут сохранены в базу данных.

Пример 2. Использование call_command с транзакцией и отловом исключений

Часто требуется вызвать несколько команд в одной транзакции. Можно использовать декоратор transaction.atomic:

from django.db import transaction
from django.core.management import call_command

@transaction.atomic
def deploy_updates():
    """Применить изменения базы данных и загрузить начальные данные"""
    call_command('migrate', '--fake-initial')
    call_command('loaddata', 'initial_data.json')
    call_command('collectstatic', '--noinput')

Если одна из команд вызовет исключение, все изменения откатятся. Полезно для скриптов развёртывания.

Результат: все команды выполнятся последовательно, при ошибке база вернётся к исходному состоянию.

Пример 3. Вывод информации о проекте через manage.py

Стандартная команда check проверяет проект на наличие проблем, но можно написать свою для вывода версий приложений или настроек.

# myapp/management/commands/project_info.py
from django.conf import settings
from django.core.management.base import BaseCommand

class Command(BaseCommand):
    help = 'Показывает информацию о проекте'

    def handle(self, *args, **options):
        self.stdout.write(f'DEBUG: {settings.DEBUG}')
        self.stdout.write(f'База данных: {settings.DATABASES["default"]["ENGINE"]}')
        self.stdout.write(f'Установленные приложения: {len(settings.INSTALLED_APPS)}')
        self.stdout.write(f'BROKER_URL: {getattr(settings, "CELERY_BROKER_URL", "не настроен")}')
        # Вывод списка URL-шаблонов (если доступны)
        try:
            from django.urls import get_resolver
            from django.urls.resolvers import URLPattern
            resolver = get_resolver()
            self.stdout.write('URL-шаблоны:')
            for pattern in resolver.url_patterns:
                if isinstance(pattern, URLPattern):
                    self.stdout.write(f'  {pattern.pattern}')
        except Exception as e:
            self.stdout.write(self.style.WARNING(f'Не удалось прочитать URL: {e}'))

Пример вызова и вывода:

python manage.py project_info

DEBUG: True
База данных: django.db.backends.sqlite3
Установленные приложения: 12
BROKER_URL: не настроен
URL-шаблоны:
  admin/
  api/

Пример 4. Запуск нескольких команд в одном процессе с передачей контекста

Иногда нужно выполнить команды, которые зависят друг от друга. Например, инициализация демо-данных после миграции.

from django.core.management import call_command, execute_from_command_line

# Прямой вызов в скрипте
def setup_demo():
    call_command('migrate')
    call_command('loaddata', 'demo_data')
    call_command('rebuild_index', verbosity=0)

# Вызов из manage.py с передачей всей строки через execute_from_command_line
# (этот метод принимает список аргументов, например, из sys.argv)
execute_from_command_line(['manage.py', 'migrate'])
execute_from_command_line(['manage.py', 'loaddata', 'demo_data'])

Второй способ полезен, когда нужно динамически формировать аргументы. Однако каждый вызов перезагружает проекты, что неэффективно. Лучше использовать call_command.

Пример 5. Кастомная команда с прогресс-баром

Для длительных операций можно отображать индикатор выполнения с помощью tqdm или встроенного self.stdout.write с перезаписью строки.

# myapp/management/commands/bulk_update_prices.py
from django.core.management.base import BaseCommand
from myapp.models import Product
import time

class Command(BaseCommand):
    help = 'Обновляет цены всех товаров с заданным коэффициентом'

    def add_arguments(self, parser):
        parser.add_argument('factor', type=float, help='Коэффициент увеличения цены')

    def handle(self, *args, **options):
        factor = options['factor']
        qs = Product.objects.all()
        total = qs.count()
        self.stdout.write(f'Обновление {total} товаров...')
        for idx, product in enumerate(qs.iterator(), 1):
            product.price = float(product.price) * factor
            product.save(update_fields=['price'])
            if idx % 10 == 0:
                self.stdout.write(f'\rОбработано {idx}/{total}', ending='')
                self.stdout.flush()
        self.stdout.write(f'\nОбновление завершено')

Вывод в консоли:
Обновление 100 товаров...
Обработано 30/100
Обработано 40/100
...
Обработано 100/100
Обновление завершено