Management Commands

class Command(BaseCommand):
    help = "Export data from a specified resource or model in a chosen format."

    def add_arguments(self, parser):
        """
        Add command-line arguments for export command.
        
        Arguments:
        - format: Export format (CSV, XLSX, JSON, etc.)
        - resource: Resource class or model class in dotted path format
        - --encoding: Character encoding for text formats
        """

    def handle(self, *args, **options):
        """
        Handle export command execution.
        
        Parameters:
        - *args: Positional arguments
        - **options: Command options from arguments
        """

class Command(BaseCommand):
    help = "Import data from a file into a specified resource or model."

    def add_arguments(self, parser):
        """
        Add command-line arguments for import command.
        
        Arguments:
        - format: Import format (CSV, XLSX, JSON, etc.)
        - resource: Resource class or model class in dotted path format
        - file: Path to file to import
        - --encoding: Character encoding for text formats
        - --dry-run: Perform dry run without saving data
        """

    def handle(self, *args, **options):
        """
        Handle import command execution.
        
        Parameters:
        - *args: Positional arguments
        - **options: Command options from arguments
        """

python manage.py export <format> <resource> [--encoding=<encoding>]

python manage.py import <format> <resource> <file> [--encoding=<encoding>] [--dry-run]

# Export books to CSV format
python manage.py export CSV myapp.models.Book

# Export using custom resource
python manage.py export CSV myapp.resources.BookResource

# Export to Excel format
python manage.py export XLSX myapp.models.Book

# Export to JSON format
python manage.py export JSON myapp.models.Book

# Export with specific encoding
python manage.py export CSV myapp.models.Book --encoding=latin-1

# Import books from CSV file
python manage.py import CSV myapp.models.Book books.csv

# Import using custom resource
python manage.py import CSV myapp.resources.BookResource books.csv

# Import from Excel file
python manage.py import XLSX myapp.models.Book books.xlsx

# Import from JSON file
python manage.py import JSON myapp.models.Book books.json

# Dry run import (validation only)
python manage.py import CSV myapp.models.Book books.csv --dry-run

# Import with specific encoding
python manage.py import CSV myapp.models.Book books.csv --encoding=utf-8

# Export with custom format class
python manage.py export myproject.formats.CustomCSV myapp.models.Book

# Import large dataset
python manage.py import CSV myapp.resources.BulkBookResource large_books.csv

# Import with error handling
python manage.py import CSV myapp.models.Book books.csv --dry-run > validation_report.txt

# Chain commands for processing
python manage.py export CSV myapp.models.Book > temp_export.csv
python manage.py import CSV myapp.models.Book temp_export.csv --dry-run

#!/bin/bash
# Export and backup script

DATE=$(date +%Y%m%d)
BACKUP_DIR="/backups/django_exports/$DATE"
mkdir -p "$BACKUP_DIR"

# Export all models
python manage.py export CSV myapp.models.Book > "$BACKUP_DIR/books.csv"
python manage.py export CSV myapp.models.Author > "$BACKUP_DIR/authors.csv"
python manage.py export CSV myapp.models.Publisher > "$BACKUP_DIR/publishers.csv"

echo "Export completed to $BACKUP_DIR"

#!/bin/bash
# Import validation script

IMPORT_FILE="$1"
RESOURCE="$2"

if [ -z "$IMPORT_FILE" ] || [ -z "$RESOURCE" ]; then
    echo "Usage: $0 <import_file> <resource>"
    exit 1
fi

echo "Validating import file: $IMPORT_FILE"
python manage.py import CSV "$RESOURCE" "$IMPORT_FILE" --dry-run

if [ $? -eq 0 ]; then
    echo "Validation passed. Proceeding with import..."
    python manage.py import CSV "$RESOURCE" "$IMPORT_FILE"
else
    echo "Validation failed. Import aborted."
    exit 1
fi

from django.core.management.base import BaseCommand, CommandError
from import_export.command_utils import get_resource_class, get_format_class
import sys
import os

class Command(BaseCommand):
    help = 'Enhanced export command with additional options'

    def add_arguments(self, parser):
        parser.add_argument('format', help='Export format')
        parser.add_argument('resource', help='Resource or model class')
        parser.add_argument(
            '--output', '-o',
            help='Output file path (default: stdout)'
        )
        parser.add_argument(
            '--encoding',
            default='utf-8',
            help='Character encoding (default: utf-8)'
        )
        parser.add_argument(
            '--limit',
            type=int,
            help='Limit number of records to export'
        )
        parser.add_argument(
            '--filter',
            help='Django filter expression (e.g., "published=True")'
        )
        parser.add_argument(
            '--fields',
            help='Comma-separated list of fields to export'
        )

    def handle(self, *args, **options):
        try:
            # Get resource and format
            resource_class = get_resource_class(options['resource'])
            format_class = get_format_class(options['format'])
            
            # Initialize resource
            resource = resource_class()
            
            # Get queryset
            if hasattr(resource, 'get_queryset'):
                queryset = resource.get_queryset()
                
                # Apply filters
                if options['filter']:
                    filter_kwargs = {}
                    for filter_expr in options['filter'].split(','):
                        key, value = filter_expr.split('=')
                        filter_kwargs[key.strip()] = value.strip()
                    queryset = queryset.filter(**filter_kwargs)
                
                # Apply limit
                if options['limit']:
                    queryset = queryset[:options['limit']]
            else:
                queryset = None
            
            # Export data
            dataset = resource.export(queryset)
            
            # Filter fields if specified
            if options['fields']:
                field_names = [f.strip() for f in options['fields'].split(',')]
                # Filter dataset to only include specified fields
                headers = dataset.headers
                field_indices = [headers.index(f) for f in field_names if f in headers]
                
                filtered_data = []
                for row in dataset:
                    filtered_row = [row[i] for i in field_indices]
                    filtered_data.append(filtered_row)
                
                dataset.headers = [headers[i] for i in field_indices]
                dataset._dataset = filtered_data
            
            # Export to format
            export_data = format_class.export_data(dataset)
            
            # Output data
            if options['output']:
                mode = 'wb' if format_class.is_binary() else 'w'
                encoding = None if format_class.is_binary() else options['encoding']
                
                with open(options['output'], mode, encoding=encoding) as f:
                    f.write(export_data)
                
                self.stdout.write(
                    self.style.SUCCESS(f'Exported to {options["output"]}')
                )
            else:
                if format_class.is_binary():
                    sys.stdout.buffer.write(export_data)
                else:
                    self.stdout.write(export_data)
                    
        except Exception as e:
            raise CommandError(f'Export failed: {e}')

from django.core.management.base import BaseCommand, CommandError
from import_export.command_utils import get_resource_class, get_format_class
from import_export.results import Result
import os

class Command(BaseCommand):
    help = 'Enhanced import command with detailed reporting'

    def add_arguments(self, parser):
        parser.add_argument('format', help='Import format')
        parser.add_argument('resource', help='Resource or model class')
        parser.add_argument('file', help='File to import')
        parser.add_argument(
            '--encoding',
            default='utf-8',
            help='Character encoding (default: utf-8)'
        )
        parser.add_argument(
            '--dry-run',
            action='store_true',
            help='Perform dry run without saving'
        )
        parser.add_argument(
            '--batch-size',
            type=int,
            default=1000,
            help='Batch size for bulk operations'
        )
        parser.add_argument(
            '--skip-errors',
            action='store_true',
            help='Skip rows with errors and continue'
        )
        parser.add_argument(
            '--report',
            help='Path to save detailed import report'
        )

    def handle(self, *args, **options):
        try:
            # Validate file exists
            if not os.path.exists(options['file']):
                raise CommandError(f'File not found: {options["file"]}')
            
            # Get resource and format
            resource_class = get_resource_class(options['resource'])
            format_class = get_format_class(options['format'])
            
            # Initialize resource
            resource = resource_class()
            
            # Read and parse file
            mode = 'rb' if format_class.is_binary() else 'r'
            encoding = None if format_class.is_binary() else options['encoding']
            
            with open(options['file'], mode, encoding=encoding) as f:
                content = f.read()
                dataset = format_class.create_dataset(content)
            
            self.stdout.write(f'Loaded {len(dataset)} rows from {options["file"]}')
            
            # Perform import
            result = resource.import_data(
                dataset,
                dry_run=options['dry_run'],
                raise_errors=not options['skip_errors'],
                use_transactions=True,
                collect_failed_rows=True,
            )
            
            # Report results
            self.report_results(result, options)
            
            # Save detailed report if requested
            if options['report']:
                self.save_report(result, options['report'])
                
        except Exception as e:
            raise CommandError(f'Import failed: {e}')

    def report_results(self, result, options):
        """Print import results summary."""
        
        if options['dry_run']:
            self.stdout.write(self.style.WARNING('DRY RUN - No data was saved'))
        
        self.stdout.write(f'Total rows processed: {result.total_rows}')
        self.stdout.write(f'Valid rows: {len(result.valid_rows())}')
        
        if result.has_errors():
            self.stdout.write(
                self.style.ERROR(f'Errors: {len(result.base_errors)}')
            )
            for error in result.base_errors[:5]:  # Show first 5 errors
                self.stdout.write(f'  - {error}')
            
            if len(result.base_errors) > 5:
                self.stdout.write(f'  ... and {len(result.base_errors) - 5} more errors')
        
        if result.has_validation_errors():
            self.stdout.write(
                self.style.ERROR(f'Validation errors: {len(result.invalid_rows)}')
            )
        
        # Show summary by import type
        totals = result.totals
        if totals:
            self.stdout.write('Import summary:')
            for import_type, count in totals.items():
                if count > 0:
                    self.stdout.write(f'  {import_type}: {count}')

    def save_report(self, result, report_path):
        """Save detailed import report to file."""
        
        with open(report_path, 'w') as f:
            f.write('Django Import Export - Detailed Report\n')
            f.write('=' * 50 + '\n\n')
            
            f.write(f'Total rows: {result.total_rows}\n')
            f.write(f'Valid rows: {len(result.valid_rows())}\n')
            f.write(f'Error rows: {len(result.base_errors)}\n')
            f.write(f'Invalid rows: {len(result.invalid_rows)}\n\n')
            
            # Detailed errors
            if result.base_errors:
                f.write('ERRORS:\n')
                f.write('-' * 20 + '\n')
                for i, error in enumerate(result.base_errors, 1):
                    f.write(f'{i}. {error}\n')
                f.write('\n')
            
            # Validation errors
            if result.invalid_rows:
                f.write('VALIDATION ERRORS:\n')
                f.write('-' * 20 + '\n')
                for i, invalid_row in enumerate(result.invalid_rows, 1):
                    f.write(f'{i}. Row {invalid_row.number}: {invalid_row.error}\n')
                f.write('\n')
            
            # Success summary
            if result.totals:
                f.write('SUMMARY:\n')
                f.write('-' * 20 + '\n')
                for import_type, count in result.totals.items():
                    f.write(f'{import_type}: {count}\n')
        
        self.stdout.write(f'Detailed report saved to: {report_path}')

class Command(BaseCommand):
    help = 'Batch process multiple import files'

    def add_arguments(self, parser):
        parser.add_argument('format', help='Import format')
        parser.add_argument('resource', help='Resource or model class')
        parser.add_argument('directory', help='Directory containing files to import')
        parser.add_argument(
            '--pattern',
            default='*',
            help='File pattern to match (default: *)'
        )
        parser.add_argument(
            '--dry-run',
            action='store_true',
            help='Perform dry run without saving'
        )

    def handle(self, *args, **options):
        import glob
        
        # Find files to process
        pattern = os.path.join(options['directory'], options['pattern'])
        files = glob.glob(pattern)
        
        if not files:
            self.stdout.write(f'No files found matching pattern: {pattern}')
            return
        
        self.stdout.write(f'Found {len(files)} files to process')
        
        # Process each file
        total_processed = 0
        total_errors = 0
        
        for file_path in files:
            self.stdout.write(f'\nProcessing: {os.path.basename(file_path)}')
            
            try:
                # Import file using the enhanced import logic
                result = self.import_file(file_path, options)
                
                total_processed += result.total_rows
                if result.has_errors():
                    total_errors += len(result.base_errors)
                    
                self.stdout.write(
                    f'  Processed {result.total_rows} rows, '
                    f'{len(result.base_errors)} errors'
                )
                
            except Exception as e:
                self.stdout.write(
                    self.style.ERROR(f'  Failed to process file: {e}')
                )
        
        # Final summary
        self.stdout.write(f'\nBatch processing complete:')
        self.stdout.write(f'  Files processed: {len(files)}')
        self.stdout.write(f'  Total rows: {total_processed}')
        self.stdout.write(f'  Total errors: {total_errors}')

    def import_file(self, file_path, options):
        """Import a single file and return results."""
        
        resource_class = get_resource_class(options['resource'])
        format_class = get_format_class(options['format'])
        
        resource = resource_class()
        
        with open(file_path, 'r') as f:
            dataset = format_class.create_dataset(f.read())
        
        return resource.import_data(
            dataset,
            dry_run=options['dry_run'],
            raise_errors=False,
        )

# Dockerfile for import/export operations
FROM python:3.9
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .

# Command for data export
CMD ["python", "manage.py", "export", "CSV", "myapp.models.Book"]

name: Data Import/Export

on:
  schedule:
    - cron: '0 2 * * *'  # Daily at 2 AM
  workflow_dispatch:

jobs:
  export-data:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v2
      
      - name: Set up Python
        uses: actions/setup-python@v2
        with:
          python-version: 3.9
          
      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          
      - name: Export data
        run: |
          python manage.py export CSV myapp.models.Book > books_export.csv
          python manage.py export CSV myapp.models.Author > authors_export.csv
          
      - name: Upload exports
        uses: actions/upload-artifact@v2
        with:
          name: data-exports
          path: |
            books_export.csv
            authors_export.csv

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

logger = logging.getLogger('import_export.commands')

class Command(BaseCommand):
    def handle(self, *args, **options):
        logger.info(f'Starting export: {options}')
        
        try:
            # Export logic here
            result = self.perform_export(options)
            logger.info(f'Export completed successfully: {result}')
            
        except Exception as e:
            logger.error(f'Export failed: {e}', exc_info=True)
            raise