Management Commands¶

The winidjango.core.commands package provides a powerful framework for building Django management commands with built-in best practices, automatic logging, and standardized argument handling.

Table of Contents¶

ABCBaseCommand
ImportDataBaseCommand
Built-in Arguments
Best Practices
Examples

ABCBaseCommand¶

Module: winidjango.core.commands.base.base

Abstract base class that provides a robust foundation for all Django management commands.

Overview¶

ABCBaseCommand implements the template method pattern to enforce consistent command structure while allowing customization. It combines Django's BaseCommand with automatic logging and standardized argument handling.

Key Features:

Template method pattern for consistent execution flow
Automatic logging via ABCLoggingMixin from winiutils
Built-in common arguments (dry-run, batch size, threading, etc.)
Type-safe with full type hints
Abstract method enforcement at compile-time

Class Structure¶

from winidjango.core.commands.base.base import ABCBaseCommand
from argparse import ArgumentParser

class MyCommand(ABCBaseCommand):
    """Your custom command."""

    def add_command_arguments(self, parser: ArgumentParser) -> None:
        """Add command-specific arguments."""
        # Implement your custom arguments
        pass

    def handle_command(self) -> None:
        """Execute command logic."""
        # Implement your command logic
        pass

Abstract Methods¶

`add_command_arguments()`¶

Add command-specific arguments to the argument parser.

Signature:

@abstractmethod
def add_command_arguments(self, parser: ArgumentParser) -> None:
    ...

Parameters:

parser: Django's ArgumentParser instance

Example:

def add_command_arguments(self, parser: ArgumentParser) -> None:
    parser.add_argument(
        '--input-file',
        type=str,
        required=True,
        help='Path to input file'
    )
    parser.add_argument(
        '--output-format',
        choices=['json', 'csv', 'xml'],
        default='json',
        help='Output format for results'
    )
    parser.add_argument(
        '--limit',
        type=int,
        help='Maximum number of records to process'
    )

`handle_command()`¶

Execute the command-specific logic.

Signature:

@abstractmethod
def handle_command(self) -> None:
    ...

Parameters:

None (access arguments via self.get_option())

Example:

def handle_command(self) -> None:
    # Get command-specific arguments
    input_file = self.get_option('input_file')
    output_format = self.get_option('output_format')
    limit = self.get_option('limit')

    # Get built-in arguments
    dry_run = self.get_option('dry_run')
    batch_size = self.get_option('batch_size') or 1000

    if dry_run:
        self.stdout.write(self.style.WARNING('DRY RUN MODE'))

    # Your command logic
    data = self.load_data(input_file, limit)
    self.process_data(data, batch_size, dry_run)
    self.export_results(output_format)

Methods¶

`get_option()`¶

Get an option value from parsed command-line arguments.

Signature:

def get_option(self, option: str) -> Any:
    ...

Parameters:

option: Option name (use underscores, not hyphens)

Returns:

Option value (type depends on argument definition)

Example:

def handle_command(self) -> None:
    # Command-line: --input-file data.csv
    # Access as: input_file (underscores)
    input_file = self.get_option('input_file')

    # Built-in options
    dry_run = self.get_option('dry_run')  # bool
    batch_size = self.get_option('batch_size')  # int | None
    threads = self.get_option('threads')  # int | None

Execution Flow¶

The command execution follows this flow:

handle() - Entry point (final, cannot override)
base_handle() - Common setup (stores args and options)
handle_command() - Your custom logic (abstract, must implement)

# Automatic flow:
handle()
  ├─> base_handle(*args, **options)  # Stores self.args, self.options
  └─> handle_command()                # Your implementation

Built-in Arguments¶

All commands inheriting from ABCBaseCommand automatically get these arguments:

Argument	Type	Default	Description
`--dry_run`	bool	False	Preview changes without executing
`--force`	bool	False	Force execution of actions
`--delete`	bool	False	Enable deletion operations
`--yes`	bool	False	Auto-confirm all prompts
`--timeout`	int	None	Command timeout in seconds
`--batch_size`	int	None	Batch processing size
`--threads`	int	None	Thread count for parallel processing
`--processes`	int	None	Process count for multiprocessing

Access via Options class:

class MyCommand(ABCBaseCommand):
    def handle_command(self) -> None:
        # Using Options class constants
        dry_run = self.get_option(self.Options.DRY_RUN)
        batch_size = self.get_option(self.Options.BATCH_SIZE)
        threads = self.get_option(self.Options.THREADS)

Complete Example¶

from winidjango.core.commands.base.base import ABCBaseCommand
from argparse import ArgumentParser
from winidjango.core.db.bulk import bulk_create_in_steps

class ImportProductsCommand(ABCBaseCommand):
    """Import products from CSV file."""

    def add_command_arguments(self, parser: ArgumentParser) -> None:
        parser.add_argument(
            '--file',
            type=str,
            required=True,
            help='Path to CSV file'
        )
        parser.add_argument(
            '--category',
            type=str,
            help='Filter by category'
        )

    def handle_command(self) -> None:
        file_path = self.get_option('file')
        category = self.get_option('category')
        dry_run = self.get_option('dry_run')
        batch_size = self.get_option('batch_size') or 1000

        self.stdout.write(f'Processing {file_path}')

        # Load data
        products = self.load_products(file_path, category)
        self.stdout.write(f'Loaded {len(products)} products')

        if dry_run:
            self.stdout.write(self.style.WARNING('DRY RUN - No changes made'))
            return

        # Create in database
        created = bulk_create_in_steps(Product, products, step=batch_size)
        self.stdout.write(
            self.style.SUCCESS(f'Created {len(created)} products')
        )

    def load_products(
        self, file_path: str, category: str | None
    ) -> list[Product]:
        # Your loading logic
        pass

Usage:

# Dry run
python manage.py import_products --file products.csv --dry_run

# With batch size
python manage.py import_products --file products.csv --batch_size 500

# With category filter
python manage.py import_products --file products.csv --category electronics

# Force mode
python manage.py import_products --file products.csv --force

ImportDataBaseCommand¶

Module: winidjango.core.commands.import_data

Specialized command for structured data import workflows with automatic cleaning and bulk creation.

Overview¶

ImportDataBaseCommand extends ABCBaseCommand to provide a standardized 4-step workflow for data imports:

Import - Fetch raw data from source
Clean - Apply data cleaning rules
Transform - Convert to Django model instances
Load - Bulk create with dependency resolution

Key Features:

Polars DataFrame integration for high-performance data processing
Automatic data cleaning via winiutils.CleaningDF
Dependency-aware bulk creation with topological sorting
Inherits all ABCBaseCommand features (logging, built-in args, etc.)

Abstract Methods¶

`handle_import()`¶

Fetch raw data from the source.

Signature:

@abstractmethod
def handle_import(self) -> pl.DataFrame:
    ...

Returns:

Polars DataFrame containing raw data

Example:

import polars as pl

def handle_import(self) -> pl.DataFrame:
    file_path = self.get_option('file')

    # From CSV
    return pl.read_csv(file_path)

    # From JSON
    # return pl.read_json(file_path)

    # From database
    # return pl.read_database(query, connection)

    # From API
    # data = requests.get(api_url).json()
    # return pl.DataFrame(data)

`get_cleaning_df_cls()`¶

Return the data cleaning class.

Signature:

@abstractmethod
def get_cleaning_df_cls(self) -> type[CleaningDF]:
    ...

Returns:

Subclass of winiutils.CleaningDF

Example:

from winiutils.core.data.dataframe.cleaning import CleaningDF
import polars as pl

class UserCleaningDF(CleaningDF):
    """Data cleaning rules for user import."""

    USERNAME_COL = "username"
    EMAIL_COL = "email"
    AGE_COL = "age"

    @classmethod
    def get_rename_map(cls) -> dict[str, str]:
        """Rename columns."""
        return {
            "user_name": cls.USERNAME_COL,
            "user_email": cls.EMAIL_COL,
        }

    @classmethod
    def get_col_dtype_map(cls) -> dict[str, type[pl.DataType]]:
        """Define column types."""
        return {
            cls.USERNAME_COL: pl.Utf8,
            cls.EMAIL_COL: pl.Utf8,
            cls.AGE_COL: pl.Int64,
        }

    @classmethod
    def get_no_null_cols(cls) -> tuple[str, ...]:
        """Columns that cannot be null."""
        return (cls.USERNAME_COL, cls.EMAIL_COL)

    @classmethod
    def get_unique_subsets(cls) -> tuple[tuple[str, ...], ...]:
        """Define uniqueness constraints."""
        return ((cls.USERNAME_COL,), (cls.EMAIL_COL,))

def get_cleaning_df_cls(self) -> type[CleaningDF]:
    return UserCleaningDF

`get_bulks_by_model()`¶

Convert cleaned DataFrame to Django model instances.

Signature:

@abstractmethod
def get_bulks_by_model(
    self, df: pl.DataFrame
) -> dict[type[Model], Iterable[Model]]:
    ...

Parameters:

df: Cleaned Polars DataFrame

Returns:

Dictionary mapping model classes to iterables of instances

Example:

def get_bulks_by_model(
    self, df: pl.DataFrame
) -> dict[type[Model], Iterable[Model]]:
    # Convert DataFrame rows to model instances
    users = [
        User(
            username=row["username"],
            email=row["email"],
            age=row["age"]
        )
        for row in df.iter_rows(named=True)
    ]

    # Create related models
    profiles = [Profile(user=user) for user in users]

    # Return in any order - automatic dependency resolution
    return {
        User: users,
        Profile: profiles,  # Created after User
    }

Methods¶

`import_to_db()`¶

Import cleaned data to database (called automatically).

Signature:

def import_to_db(self) -> None:
    ...

Behavior:

Calls get_bulks_by_model() to get model instances
Uses bulk_create_bulks_in_steps() for dependency-aware creation
Automatically sorts models by foreign key dependencies

Note: You typically don't override this method.

Complete Example¶

from winidjango.core.commands.import_data import ImportDataBaseCommand
from winiutils.core.data.dataframe.cleaning import CleaningDF
from argparse import ArgumentParser
import polars as pl

class ProductCleaningDF(CleaningDF):
    """Cleaning rules for product data."""

    NAME_COL = "name"
    PRICE_COL = "price"
    SKU_COL = "sku"

    @classmethod
    def get_col_dtype_map(cls) -> dict[str, type[pl.DataType]]:
        return {
            cls.NAME_COL: pl.Utf8,
            cls.PRICE_COL: pl.Float64,
            cls.SKU_COL: pl.Utf8,
        }

    @classmethod
    def get_no_null_cols(cls) -> tuple[str, ...]:
        return (cls.NAME_COL, cls.SKU_COL)

    @classmethod
    def get_unique_subsets(cls) -> tuple[tuple[str, ...], ...]:
        return ((cls.SKU_COL,),)

class ImportProductsCommand(ImportDataBaseCommand):
    """Import products from CSV with automatic cleaning."""

    def add_command_arguments(self, parser: ArgumentParser) -> None:
        parser.add_argument('--file', type=str, required=True)
        parser.add_argument('--category-id', type=int, required=True)

    def handle_import(self) -> pl.DataFrame:
        file_path = self.get_option('file')
        return pl.read_csv(file_path)

    def get_cleaning_df_cls(self) -> type[CleaningDF]:
        return ProductCleaningDF

    def get_bulks_by_model(
        self, df: pl.DataFrame
    ) -> dict[type[Model], Iterable[Model]]:
        category_id = self.get_option('category_id')

        products = [
            Product(
                name=row["name"],
                price=row["price"],
                sku=row["sku"],
                category_id=category_id
            )
            for row in df.iter_rows(named=True)
        ]

        return {Product: products}

Usage:

# Import products
python manage.py import_products --file products.csv --category-id 5

# Dry run
python manage.py import_products --file products.csv --category-id 5 --dry_run

Built-in Arguments¶

Detailed Reference¶

`--dry_run`¶

Preview changes without executing them.

Type: Boolean flag Default: False

Example:

def handle_command(self) -> None:
    dry_run = self.get_option('dry_run')

    if dry_run:
        self.stdout.write(self.style.WARNING(
            'DRY RUN MODE - No changes will be made'
        ))
        # Show what would happen
        self.preview_changes()
        return

    # Actual execution
    self.execute_changes()

Usage:

python manage.py mycommand --dry_run

`--batch_size`¶

Configure batch processing size.

Type: Integer Default: None (use command's default)

Example:

def handle_command(self) -> None:
    batch_size = self.get_option('batch_size') or 1000

    bulk_create_in_steps(Model, objects, step=batch_size)

Usage:

python manage.py mycommand --batch_size 500

`--threads`¶

Control thread count for parallel processing.

Type: Integer Default: None (use system default)

Example:

def handle_command(self) -> None:
    threads = self.get_option('threads') or 4

    with ThreadPoolExecutor(max_workers=threads) as executor:
        executor.map(process_item, items)

Usage:

python manage.py mycommand --threads 8

`--force`¶

Force execution of actions (skip confirmations).

Type: Boolean flag Default: False

Example:

def handle_command(self) -> None:
    force = self.get_option('force')

    if not force:
        confirm = input('Are you sure? (y/n): ')
        if confirm.lower() != 'y':
            self.stdout.write('Cancelled')
            return

    self.execute_dangerous_operation()

Usage:

python manage.py mycommand --force

`--delete`¶

Enable deletion operations.

Type: Boolean flag Default: False

Example:

def handle_command(self) -> None:
    delete = self.get_option('delete')

    if delete:
        self.delete_old_records()
    else:
        self.stdout.write('Skipping deletion (use --delete to enable)')

Usage:

python manage.py mycommand --delete

`--yes`¶

Auto-confirm all prompts.

Type: Boolean flag Default: False

Example:

def handle_command(self) -> None:
    yes = self.get_option('yes')

    if not yes:
        confirm = input('Proceed? (y/n): ')
        if confirm.lower() != 'y':
            return

    self.execute()

Usage:

python manage.py mycommand --yes

`--timeout`¶

Set command timeout in seconds.

Type: Integer Default: None (no timeout)

Example:

import signal

def handle_command(self) -> None:
    timeout = self.get_option('timeout')

    if timeout:
        signal.alarm(timeout)

    try:
        self.long_running_operation()
    except TimeoutError:
        self.stdout.write(self.style.ERROR('Command timed out'))

Usage:

python manage.py mycommand --timeout 300  # 5 minutes

`--processes`¶

Control process count for multiprocessing.

Type: Integer Default: None (use system default)

Example:

from multiprocessing import Pool

def handle_command(self) -> None:
    processes = self.get_option('processes') or 4

    with Pool(processes=processes) as pool:
        results = pool.map(process_item, items)

Usage:

python manage.py mycommand --processes 4

Best Practices¶

1. Use Dry Run for Destructive Operations¶

def handle_command(self) -> None:
    dry_run = self.get_option('dry_run')

    # Preview deletions
    if dry_run:
        preview = simulate_bulk_deletion(Model, objects)
        self.stdout.write(f'Would delete {len(preview)} objects')
        return

    # Actual deletion
    bulk_delete_in_steps(Model, objects)

2. Provide Progress Feedback¶

def handle_command(self) -> None:
    total = len(items)

    for i, item in enumerate(items, 1):
        self.process_item(item)

        if i % 100 == 0:
            self.stdout.write(f'Processed {i}/{total} items')

    self.stdout.write(self.style.SUCCESS(f'Completed {total} items'))

3. Handle Errors Gracefully¶

def handle_command(self) -> None:
    errors = []

    for item in items:
        try:
            self.process_item(item)
        except Exception as e:
            errors.append((item, str(e)))
            self.stdout.write(self.style.ERROR(f'Error processing {item}: {e}'))

    if errors:
        self.stdout.write(self.style.WARNING(f'{len(errors)} errors occurred'))
    else:
        self.stdout.write(
            self.style.SUCCESS('All items processed successfully')
        )

4. Use Batch Size Appropriately¶

def handle_command(self) -> None:
    batch_size = self.get_option('batch_size')

    # Adjust based on object complexity
    if not batch_size:
        if self.is_complex_model():
            batch_size = 100
        else:
            batch_size = 1000

    bulk_create_in_steps(Model, objects, step=batch_size)

5. Leverage Built-in Arguments¶

def handle_command(self) -> None:
    # Combine multiple built-in arguments
    dry_run = self.get_option('dry_run')
    force = self.get_option('force')
    batch_size = self.get_option('batch_size') or 1000

    if dry_run:
        self.preview()
        return

    if not force and not self.confirm():
        return

    self.execute(batch_size)

6. Document Your Commands¶

class MyCommand(ABCBaseCommand):
    """
    Import users from CSV file.

    This command imports users from a CSV file with the following columns:
    - username: User's username (required)
    - email: User's email (required)
    - first_name: User's first name (optional)
    - last_name: User's last name (optional)

    Examples:
        # Basic import
        python manage.py import_users --file users.csv

        # Dry run
        python manage.py import_users --file users.csv --dry_run

        # With custom batch size
        python manage.py import_users --file users.csv --batch_size 500
    """

    help = "Import users from CSV file"

Examples¶

Example 1: Simple Data Processing Command¶

from winidjango.core.commands.base.base import ABCBaseCommand
from argparse import ArgumentParser

class CleanupOldRecordsCommand(ABCBaseCommand):
    """Delete records older than specified days."""

    help = "Delete old records from database"

    def add_command_arguments(self, parser: ArgumentParser) -> None:
        parser.add_argument(
            '--days',
            type=int,
            default=30,
            help='Delete records older than this many days'
        )
        parser.add_argument(
            '--model',
            type=str,
            required=True,
            help='Model name to clean up'
        )

    def handle_command(self) -> None:
        from django.apps import apps
        from datetime import datetime, timedelta

        days = self.get_option('days')
        model_name = self.get_option('model')
        dry_run = self.get_option('dry_run')
        delete = self.get_option('delete')

        # Get model
        model = apps.get_model(model_name)

        # Calculate cutoff date
        cutoff = datetime.now() - timedelta(days=days)

        # Get old records
        old_records = model.objects.filter(created_at__lt=cutoff)
        count = old_records.count()

        self.stdout.write(f'Found {count} records older than {days} days')

        if not delete:
            self.stdout.write(
                self.style.WARNING('Use --delete to actually delete records')
            )
            return

        if dry_run:
            self.stdout.write(
                self.style.WARNING(f'DRY RUN: Would delete {count} records')
            )
            return

        # Delete
        deleted, _ = old_records.delete()
        self.stdout.write(self.style.SUCCESS(f'Deleted {deleted} records'))

Example 2: Data Import with Validation¶

from winidjango.core.commands.import_data import ImportDataBaseCommand
from winiutils.core.data.dataframe.cleaning import CleaningDF
from argparse import ArgumentParser
import polars as pl

class OrderCleaningDF(CleaningDF):
    ORDER_ID_COL = "order_id"
    CUSTOMER_ID_COL = "customer_id"
    AMOUNT_COL = "amount"

    @classmethod
    def get_col_dtype_map(cls) -> dict[str, type[pl.DataType]]:
        return {
            cls.ORDER_ID_COL: pl.Utf8,
            cls.CUSTOMER_ID_COL: pl.Int64,
            cls.AMOUNT_COL: pl.Float64,
        }

    @classmethod
    def get_no_null_cols(cls) -> tuple[str, ...]:
        return (cls.ORDER_ID_COL, cls.CUSTOMER_ID_COL, cls.AMOUNT_COL)

class ImportOrdersCommand(ImportDataBaseCommand):
    """Import orders with customer validation."""

    def add_command_arguments(self, parser: ArgumentParser) -> None:
        parser.add_argument('--file', type=str, required=True)
        parser.add_argument('--validate-customers', action='store_true')

    def handle_import(self) -> pl.DataFrame:
        return pl.read_csv(self.get_option('file'))

    def get_cleaning_df_cls(self) -> type[CleaningDF]:
        return OrderCleaningDF

    def get_bulks_by_model(
        self, df: pl.DataFrame
    ) -> dict[type[Model], Iterable[Model]]:
        validate = self.get_option('validate_customers')

        # Validate customers exist
        if validate:
            customer_ids = df['customer_id'].to_list()
            existing = set(Customer.objects.filter(
                id__in=customer_ids
            ).values_list('id', flat=True))

            missing = set(customer_ids) - existing
            if missing:
                self.stdout.write(
                    self.style.ERROR(f'Missing customers: {missing}')
                )
                return {}

        # Create orders
        orders = [
            Order(
                order_id=row["order_id"],
                customer_id=row["customer_id"],
                amount=row["amount"]
            )
            for row in df.iter_rows(named=True)
        ]

        return {Order: orders}

Example 3: Multi-Model Import¶

from winidjango.core.commands.import_data import ImportDataBaseCommand
from winiutils.core.data.dataframe.cleaning import CleaningDF
import polars as pl

class StoreDataCleaningDF(CleaningDF):
    STORE_NAME_COL = "store_name"
    PRODUCT_NAME_COL = "product_name"
    PRICE_COL = "price"

    @classmethod
    def get_col_dtype_map(cls) -> dict[str, type[pl.DataType]]:
        return {
            cls.STORE_NAME_COL: pl.Utf8,
            cls.PRODUCT_NAME_COL: pl.Utf8,
            cls.PRICE_COL: pl.Float64,
        }

class ImportStoreDataCommand(ImportDataBaseCommand):
    """Import stores and their products."""

    def add_command_arguments(self, parser: ArgumentParser) -> None:
        parser.add_argument('--file', type=str, required=True)

    def handle_import(self) -> pl.DataFrame:
        return pl.read_csv(self.get_option('file'))

    def get_cleaning_df_cls(self) -> type[CleaningDF]:
        return StoreDataCleaningDF

    def get_bulks_by_model(
        self, df: pl.DataFrame
    ) -> dict[type[Model], Iterable[Model]]:
        # Get unique stores
        store_names = df['store_name'].unique().to_list()
        stores = [Store(name=name) for name in store_names]

        # Create store name to instance mapping
        store_map = {store.name: store for store in stores}

        # Create products
        products = [
            Product(
                name=row["product_name"],
                price=row["price"],
                store=store_map[row["store_name"]]
            )
            for row in df.iter_rows(named=True)
        ]

        # Return both - automatic dependency resolution
        return {
            Store: stores,
            Product: products,  # Created after Store
        }

Management Commands¶

Table of Contents¶

ABCBaseCommand¶

Overview¶

Class Structure¶

Abstract Methods¶

add_command_arguments()¶

handle_command()¶

Methods¶

get_option()¶

Execution Flow¶

Built-in Arguments¶

Complete Example¶

ImportDataBaseCommand¶

Overview¶

Abstract Methods¶

handle_import()¶

get_cleaning_df_cls()¶

get_bulks_by_model()¶

Methods¶

import_to_db()¶

Complete Example¶

Built-in Arguments¶

Detailed Reference¶

--dry_run¶

--batch_size¶

--threads¶

--force¶

--delete¶

--yes¶

--timeout¶

--processes¶

Best Practices¶

1. Use Dry Run for Destructive Operations¶

2. Provide Progress Feedback¶

3. Handle Errors Gracefully¶

4. Use Batch Size Appropriately¶

5. Leverage Built-in Arguments¶

6. Document Your Commands¶

Examples¶

Example 1: Simple Data Processing Command¶

Example 2: Data Import with Validation¶

Example 3: Multi-Model Import¶

See Also¶

`add_command_arguments()`¶

`handle_command()`¶

`get_option()`¶

`handle_import()`¶

`get_cleaning_df_cls()`¶

`get_bulks_by_model()`¶

`import_to_db()`¶

`--dry_run`¶

`--batch_size`¶

`--threads`¶

`--force`¶

`--delete`¶

`--yes`¶

`--timeout`¶

`--processes`¶