Skip to content

API Reference

init module.

in_this_repo = Path(winidjango.__name__).exists() module-attribute

installed_apps = ['tests'] if Path('tests').exists() else [] module-attribute

logger = logging.getLogger(__name__) module-attribute

rig

init module.

resources

init module.

tools

Tool wrappers for CLI tools used in development workflows.

Tools are subclasses of Tool providing methods that return Args objects for type-safe command construction and execution.

tools

Override pyrig tools.

ProjectTester

Bases: ProjectTester

Subclass of ProjectTester for customizing pyrig behavior.

Source code in winidjango/rig/tools/tools.py 
15
16
17
18
19
20
class ProjectTester(BaseProjectTester):
    """Subclass of ProjectTester for customizing pyrig behavior."""

    def dev_dependencies(self) -> tuple[str, ...]:
        """Get the dev dependencies."""
        return (*super().dev_dependencies(), "pytest-django")
dev_dependencies()

Get the dev dependencies.

Source code in winidjango/rig/tools/tools.py 
18
19
20
def dev_dependencies(self) -> tuple[str, ...]:
    """Get the dev dependencies."""
    return (*super().dev_dependencies(), "pytest-django")
Pyrigger

Bases: Pyrigger

Subclass of Pyrigger for customizing pyrig behavior.

Source code in winidjango/rig/tools/tools.py 
 7
 8
 9
10
11
12
class Pyrigger(BasePyrigger):
    """Subclass of Pyrigger for customizing pyrig behavior."""

    def dev_dependencies(self) -> tuple[str, ...]:
        """Get the dev dependencies."""
        return (*super().dev_dependencies(), "django-stubs")
dev_dependencies()

Get the dev dependencies.

Source code in winidjango/rig/tools/tools.py 
10
11
12
def dev_dependencies(self) -> tuple[str, ...]:
    """Get the dev dependencies."""
    return (*super().dev_dependencies(), "django-stubs")

src

src package.

This package exposes the project's internal modules used by the command-line utilities and database helpers. It exists primarily so that code under winidjango/src can be imported using the winidjango.src package path in other modules and tests.

The package itself contains the following subpackages: - commands - management command helpers and base classes - db - database utilities and model helpers

Consumers should import the specific submodules they need rather than relying on side effects from this package's import-time execution.

commands

Utilities and base classes for management commands.

The commands package contains base command classes and helpers used to implement Django management commands in the project. Subpackages and modules under commands provide reusable patterns for argument handling, logging and common command behaviors so that individual commands can focus on their business logic.

base

Base helpers and abstractions for management commands.

This package provides a common abstract base class used by project management commands. The base class centralizes argument handling, standard options (dry-run, batching, timeouts, etc.) and integrates logging behavior used throughout the commands package.

base

Utilities and an abstract base class for Django management commands.

This module defines :class:ABCBaseCommand, a reusable abstract base that combines Django's BaseCommand with the project's logging mixins and standard argument handling. The base class implements a template method pattern so concrete commands only need to implement the abstract extension points for providing command-specific arguments and business logic.

ABCBaseCommand

Bases: ABCLoggingMixin, BaseCommand

Abstract base class for management commands with logging and standard options.

The class wires common behavior such as base arguments (dry-run, batching, timeouts) and provides extension points that concrete commands must implement: :meth:add_command_arguments and :meth:handle_command.

Notes
  • Inheritance order matters: the logging mixin must precede BaseCommand so mixin initialization occurs as expected.
  • The base class follows the template method pattern; concrete commands should not override :meth:add_arguments or :meth:handle but implement the abstract hooks instead.
Source code in winidjango/src/commands/base/base.py 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
class ABCBaseCommand(ABCLoggingMixin, BaseCommand):
    """Abstract base class for management commands with logging and standard options.

    The class wires common behavior such as base arguments (dry-run,
    batching, timeouts) and provides extension points that concrete
    commands must implement: :meth:`add_command_arguments` and
    :meth:`handle_command`.

    Notes:
        - Inheritance order matters: the logging mixin must precede
          ``BaseCommand`` so mixin initialization occurs as expected.
        - The base class follows the template method pattern; concrete
          commands should not override :meth:`add_arguments` or
          :meth:`handle` but implement the abstract hooks instead.
    """

    class Options:
        """Just a container class for hard coding the option keys."""

        DRY_RUN = "dry_run"
        FORCE = "force"
        DELETE = "delete"
        YES = "yes"
        TIMEOUT = "timeout"
        BATCH_SIZE = "batch_size"
        THREADS = "threads"
        PROCESSES = "processes"

    def add_arguments(self, parser: ArgumentParser) -> None:
        """Configure command-line arguments for the command.

        Adds common base arguments (dry-run, force, delete, timeout,
        batching and concurrency options) and then delegates to
        :meth:`add_command_arguments` for command-specific options.

        Args:
            parser (ArgumentParser): The argument parser passed by Django.
        """
        # add base args that are used in most commands
        self.base_add_arguments(parser)

        # add additional args that are specific to the command
        self.add_command_arguments(parser)

    def base_add_arguments(self, parser: ArgumentParser) -> None:
        """Add the project's standard command-line arguments to ``parser``.

        Args:
            parser (ArgumentParser): The argument parser passed by Django.
        """
        parser.add_argument(
            f"--{self.Options.DRY_RUN}",
            action="store_true",
            help="Show what would be done without actually executing the changes",
        )

        parser.add_argument(
            f"--{self.Options.FORCE}",
            action="store_true",
            help="Force an action in a command",
        )

        parser.add_argument(
            f"--{self.Options.DELETE}",
            action="store_true",
            help="Deleting smth in a command",
        )

        parser.add_argument(
            f"--{self.Options.YES}",
            action="store_true",
            help="Answer yes to all prompts",
            default=False,
        )

        parser.add_argument(
            f"--{self.Options.TIMEOUT}",
            type=int,
            help="Timeout for a command",
            default=None,
        )

        parser.add_argument(
            f"--{self.Options.BATCH_SIZE}",
            type=int,
            default=None,
            help="Number of items to process in each batch",
        )

        parser.add_argument(
            f"--{self.Options.THREADS}",
            type=int,
            default=None,
            help="Number of threads to use for processing",
        )

        parser.add_argument(
            f"--{self.Options.PROCESSES}",
            type=int,
            default=None,
            help="Number of processes to use for processing",
        )

    @abstractmethod
    def add_command_arguments(self, parser: ArgumentParser) -> None:
        """Define command-specific arguments.

        Implement this hook to add options and positional arguments that are
        specific to the concrete management command.

        Args:
            parser (ArgumentParser): The argument parser passed by Django.
        """

    def handle(self, *args: Any, **options: Any) -> None:
        """Orchestrate command execution.

        Performs shared pre-processing by calling :meth:`base_handle` and
        then delegates to :meth:`handle_command` which must be implemented
        by subclasses.

        Args:
            *args: Positional arguments forwarded from Django.
            **options: Parsed command-line options.
        """
        self.base_handle(*args, **options)
        self.handle_command()

    def base_handle(self, *args: Any, **options: Any) -> None:
        """Perform common pre-processing for commands.

        Stores the incoming arguments and options on the instance for use
        by :meth:`handle_command` and subclasses.

        Args:
            *args: Positional arguments forwarded from Django.
            **options: Parsed command-line options.
        """
        self.args = args
        self.options = options

    @abstractmethod
    def handle_command(self) -> None:
        """Run the command-specific behavior.

        This abstract hook should be implemented by concrete commands to
        perform the command's main work. Implementations should read
        ``self.args`` and ``self.options`` which were set in
        :meth:`base_handle`.
        """

    def get_option(self, option: str) -> Any:
        """Retrieve a parsed command option by key.

        Args:
            option (str): The option key to retrieve from ``self.options``.

        Returns:
            Any: The value for the requested option. If the option is not
                present a ``KeyError`` will be raised (matching how Django
                exposes options in management commands).
        """
        return self.options[option]
Options

Just a container class for hard coding the option keys.

Source code in winidjango/src/commands/base/base.py 
38
39
40
41
42
43
44
45
46
47
48
class Options:
    """Just a container class for hard coding the option keys."""

    DRY_RUN = "dry_run"
    FORCE = "force"
    DELETE = "delete"
    YES = "yes"
    TIMEOUT = "timeout"
    BATCH_SIZE = "batch_size"
    THREADS = "threads"
    PROCESSES = "processes"
add_arguments(parser)

Configure command-line arguments for the command.

Adds common base arguments (dry-run, force, delete, timeout, batching and concurrency options) and then delegates to :meth:add_command_arguments for command-specific options.

Parameters:

Name Type Description Default
parser ArgumentParser

The argument parser passed by Django.

 required
Source code in winidjango/src/commands/base/base.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def add_arguments(self, parser: ArgumentParser) -> None:
    """Configure command-line arguments for the command.

    Adds common base arguments (dry-run, force, delete, timeout,
    batching and concurrency options) and then delegates to
    :meth:`add_command_arguments` for command-specific options.

    Args:
        parser (ArgumentParser): The argument parser passed by Django.
    """
    # add base args that are used in most commands
    self.base_add_arguments(parser)

    # add additional args that are specific to the command
    self.add_command_arguments(parser)
add_command_arguments(parser) abstractmethod

Define command-specific arguments.

Implement this hook to add options and positional arguments that are specific to the concrete management command.

Parameters:

Name Type Description Default
parser ArgumentParser

The argument parser passed by Django.

 required
Source code in winidjango/src/commands/base/base.py 
125
126
127
128
129
130
131
132
133
134
@abstractmethod
def add_command_arguments(self, parser: ArgumentParser) -> None:
    """Define command-specific arguments.

    Implement this hook to add options and positional arguments that are
    specific to the concrete management command.

    Args:
        parser (ArgumentParser): The argument parser passed by Django.
    """
base_add_arguments(parser)

Add the project's standard command-line arguments to parser.

Parameters:

Name Type Description Default
parser ArgumentParser

The argument parser passed by Django.

 required
Source code in winidjango/src/commands/base/base.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def base_add_arguments(self, parser: ArgumentParser) -> None:
    """Add the project's standard command-line arguments to ``parser``.

    Args:
        parser (ArgumentParser): The argument parser passed by Django.
    """
    parser.add_argument(
        f"--{self.Options.DRY_RUN}",
        action="store_true",
        help="Show what would be done without actually executing the changes",
    )

    parser.add_argument(
        f"--{self.Options.FORCE}",
        action="store_true",
        help="Force an action in a command",
    )

    parser.add_argument(
        f"--{self.Options.DELETE}",
        action="store_true",
        help="Deleting smth in a command",
    )

    parser.add_argument(
        f"--{self.Options.YES}",
        action="store_true",
        help="Answer yes to all prompts",
        default=False,
    )

    parser.add_argument(
        f"--{self.Options.TIMEOUT}",
        type=int,
        help="Timeout for a command",
        default=None,
    )

    parser.add_argument(
        f"--{self.Options.BATCH_SIZE}",
        type=int,
        default=None,
        help="Number of items to process in each batch",
    )

    parser.add_argument(
        f"--{self.Options.THREADS}",
        type=int,
        default=None,
        help="Number of threads to use for processing",
    )

    parser.add_argument(
        f"--{self.Options.PROCESSES}",
        type=int,
        default=None,
        help="Number of processes to use for processing",
    )
base_handle(*args, **options)

Perform common pre-processing for commands.

Stores the incoming arguments and options on the instance for use by :meth:handle_command and subclasses.

Parameters:

Name Type Description Default
*args Any

Positional arguments forwarded from Django.

 ()
**options Any

Parsed command-line options.

 {}
Source code in winidjango/src/commands/base/base.py 
150
151
152
153
154
155
156
157
158
159
160
161
def base_handle(self, *args: Any, **options: Any) -> None:
    """Perform common pre-processing for commands.

    Stores the incoming arguments and options on the instance for use
    by :meth:`handle_command` and subclasses.

    Args:
        *args: Positional arguments forwarded from Django.
        **options: Parsed command-line options.
    """
    self.args = args
    self.options = options
get_option(option)

Retrieve a parsed command option by key.

Parameters:

Name Type Description Default
option str

The option key to retrieve from self.options.

 required

Returns:

Name Type Description
Any Any

The value for the requested option. If the option is not present a KeyError will be raised (matching how Django exposes options in management commands).
Source code in winidjango/src/commands/base/base.py 
173
174
175
176
177
178
179
180
181
182
183
184
def get_option(self, option: str) -> Any:
    """Retrieve a parsed command option by key.

    Args:
        option (str): The option key to retrieve from ``self.options``.

    Returns:
        Any: The value for the requested option. If the option is not
            present a ``KeyError`` will be raised (matching how Django
            exposes options in management commands).
    """
    return self.options[option]
handle(*args, **options)

Orchestrate command execution.

Performs shared pre-processing by calling :meth:base_handle and then delegates to :meth:handle_command which must be implemented by subclasses.

Parameters:

Name Type Description Default
*args Any

Positional arguments forwarded from Django.

 ()
**options Any

Parsed command-line options.

 {}
Source code in winidjango/src/commands/base/base.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
def handle(self, *args: Any, **options: Any) -> None:
    """Orchestrate command execution.

    Performs shared pre-processing by calling :meth:`base_handle` and
    then delegates to :meth:`handle_command` which must be implemented
    by subclasses.

    Args:
        *args: Positional arguments forwarded from Django.
        **options: Parsed command-line options.
    """
    self.base_handle(*args, **options)
    self.handle_command()
handle_command() abstractmethod

Run the command-specific behavior.

This abstract hook should be implemented by concrete commands to perform the command's main work. Implementations should read self.args and self.options which were set in :meth:base_handle.

Source code in winidjango/src/commands/base/base.py 
163
164
165
166
167
168
169
170
171
@abstractmethod
def handle_command(self) -> None:
    """Run the command-specific behavior.

    This abstract hook should be implemented by concrete commands to
    perform the command's main work. Implementations should read
    ``self.args`` and ``self.options`` which were set in
    :meth:`base_handle`.
    """

import_data

Import command base class and utilities.

This module defines a reusable base command for importing tabular data into Django models. Implementations should provide a concrete source ingestion (for example, reading from CSV or an external API), a cleaning/normalization step implemented by a CleaningDF subclass, and mapping logic that groups cleaned data into model instances that can be bulk-created.

The base command centralizes the typical flow: 1. Read raw data (handle_import) 2. Wrap and clean the data using a CleaningDF subclass 3. Convert the cleaned frame into per-model bulks 4. Persist bulks using the project's bulk create helpers

Using this base class ensures a consistent import lifecycle and reduces duplicated boilerplate across different import implementations.

ImportDataBaseCommand

Bases: ABCBaseCommand

Abstract base for data-import Django management commands.

Subclasses must implement the ingestion, cleaning-class selection, and mapping of cleaned rows to Django model instances. The base implementation wires these pieces together and calls the project's bulk creation helper to persist the data.

Implementors typically only need to override the three abstract methods documented below.

Source code in winidjango/src/commands/import_data.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
class ImportDataBaseCommand(ABCBaseCommand):
    """Abstract base for data-import Django management commands.

    Subclasses must implement the ingestion, cleaning-class selection,
    and mapping of cleaned rows to Django model instances. The base
    implementation wires these pieces together and calls the project's
    bulk creation helper to persist the data.

    Implementors typically only need to override the three abstract
    methods documented below.
    """

    @abstractmethod
    def handle_import(self) -> pl.DataFrame:
        """Read raw data from the import source.

        This method should read data from whatever source the concrete
        command targets (files, remote APIs, etc.) and return it as a
        ``polars.DataFrame``. No cleaning should be performed here;
        cleaning is handled by the cleaning `CleaningDF` returned from
        ``get_cleaning_df_cls``.

        Returns:
            pl.DataFrame: Raw (uncleaned) tabular data to be cleaned and
                mapped to model instances.
        """

    @abstractmethod
    def get_cleaning_df_cls(self) -> type[CleaningDF]:
        """Return the `CleaningDF` subclass used to normalize the data.

        The returned class will be instantiated with the raw DataFrame
        returned from :meth:`handle_import` and must provide the
        transformations required to prepare data for mapping into model
        instances.

        Returns:
            type[CleaningDF]: A subclass of ``CleaningDF`` that performs
                the necessary normalization and validation.
        """

    @abstractmethod
    def get_bulks_by_model(
        self, df: pl.DataFrame
    ) -> dict[type[Model], Iterable[Model]]:
        """Map the cleaned DataFrame to model-instance bulks.

        The implementation should inspect the cleaned DataFrame and
        return a mapping where keys are Django model classes and values
        are iterables of unsaved model instances (or dataclass-like
        objects accepted by the project's bulk-creation utility).

        Args:
            df (pl.DataFrame): The cleaned and normalized DataFrame.

        Returns:
            dict[type[Model], Iterable[Model]]: Mapping from model classes
                to iterables of instances that should be created.
        """

    def handle_command(self) -> None:
        """Execute the full import lifecycle.

        This template method reads raw data via :meth:`handle_import`,
        wraps it with the cleaning class returned by
        :meth:`get_cleaning_df_cls` and then persists the resulting
        model bulks returned by :meth:`get_bulks_by_model`.
        """
        data_df = self.handle_import()

        cleaning_df_cls = self.get_cleaning_df_cls()
        self.cleaning_df = cleaning_df_cls(data_df)

        self.import_to_db()

    def import_to_db(self) -> None:
        """Persist prepared model bulks to the database.

        Calls the project's `bulk_create_bulks_in_steps` helper with the
        mapping returned from :meth:`get_bulks_by_model`.
        """
        bulks_by_model = self.get_bulks_by_model(df=self.cleaning_df.df)

        bulk_create_bulks_in_steps(bulks_by_model)
Options

Just a container class for hard coding the option keys.

Source code in winidjango/src/commands/base/base.py 
38
39
40
41
42
43
44
45
46
47
48
class Options:
    """Just a container class for hard coding the option keys."""

    DRY_RUN = "dry_run"
    FORCE = "force"
    DELETE = "delete"
    YES = "yes"
    TIMEOUT = "timeout"
    BATCH_SIZE = "batch_size"
    THREADS = "threads"
    PROCESSES = "processes"
add_arguments(parser)

Configure command-line arguments for the command.

Adds common base arguments (dry-run, force, delete, timeout, batching and concurrency options) and then delegates to :meth:add_command_arguments for command-specific options.

Parameters:

Name Type Description Default
parser ArgumentParser

The argument parser passed by Django.

 required
Source code in winidjango/src/commands/base/base.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def add_arguments(self, parser: ArgumentParser) -> None:
    """Configure command-line arguments for the command.

    Adds common base arguments (dry-run, force, delete, timeout,
    batching and concurrency options) and then delegates to
    :meth:`add_command_arguments` for command-specific options.

    Args:
        parser (ArgumentParser): The argument parser passed by Django.
    """
    # add base args that are used in most commands
    self.base_add_arguments(parser)

    # add additional args that are specific to the command
    self.add_command_arguments(parser)
add_command_arguments(parser) abstractmethod

Define command-specific arguments.

Implement this hook to add options and positional arguments that are specific to the concrete management command.

Parameters:

Name Type Description Default
parser ArgumentParser

The argument parser passed by Django.

 required
Source code in winidjango/src/commands/base/base.py 
125
126
127
128
129
130
131
132
133
134
@abstractmethod
def add_command_arguments(self, parser: ArgumentParser) -> None:
    """Define command-specific arguments.

    Implement this hook to add options and positional arguments that are
    specific to the concrete management command.

    Args:
        parser (ArgumentParser): The argument parser passed by Django.
    """
base_add_arguments(parser)

Add the project's standard command-line arguments to parser.

Parameters:

Name Type Description Default
parser ArgumentParser

The argument parser passed by Django.

 required
Source code in winidjango/src/commands/base/base.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def base_add_arguments(self, parser: ArgumentParser) -> None:
    """Add the project's standard command-line arguments to ``parser``.

    Args:
        parser (ArgumentParser): The argument parser passed by Django.
    """
    parser.add_argument(
        f"--{self.Options.DRY_RUN}",
        action="store_true",
        help="Show what would be done without actually executing the changes",
    )

    parser.add_argument(
        f"--{self.Options.FORCE}",
        action="store_true",
        help="Force an action in a command",
    )

    parser.add_argument(
        f"--{self.Options.DELETE}",
        action="store_true",
        help="Deleting smth in a command",
    )

    parser.add_argument(
        f"--{self.Options.YES}",
        action="store_true",
        help="Answer yes to all prompts",
        default=False,
    )

    parser.add_argument(
        f"--{self.Options.TIMEOUT}",
        type=int,
        help="Timeout for a command",
        default=None,
    )

    parser.add_argument(
        f"--{self.Options.BATCH_SIZE}",
        type=int,
        default=None,
        help="Number of items to process in each batch",
    )

    parser.add_argument(
        f"--{self.Options.THREADS}",
        type=int,
        default=None,
        help="Number of threads to use for processing",
    )

    parser.add_argument(
        f"--{self.Options.PROCESSES}",
        type=int,
        default=None,
        help="Number of processes to use for processing",
    )
base_handle(*args, **options)

Perform common pre-processing for commands.

Stores the incoming arguments and options on the instance for use by :meth:handle_command and subclasses.

Parameters:

Name Type Description Default
*args Any

Positional arguments forwarded from Django.

 ()
**options Any

Parsed command-line options.

 {}
Source code in winidjango/src/commands/base/base.py 
150
151
152
153
154
155
156
157
158
159
160
161
def base_handle(self, *args: Any, **options: Any) -> None:
    """Perform common pre-processing for commands.

    Stores the incoming arguments and options on the instance for use
    by :meth:`handle_command` and subclasses.

    Args:
        *args: Positional arguments forwarded from Django.
        **options: Parsed command-line options.
    """
    self.args = args
    self.options = options
get_bulks_by_model(df) abstractmethod

Map the cleaned DataFrame to model-instance bulks.

The implementation should inspect the cleaned DataFrame and return a mapping where keys are Django model classes and values are iterables of unsaved model instances (or dataclass-like objects accepted by the project's bulk-creation utility).

Parameters:

Name Type Description Default
df DataFrame

The cleaned and normalized DataFrame.

 required

Returns:

Type Description
dict[type[Model], Iterable[Model]]

dict[type[Model], Iterable[Model]]: Mapping from model classes to iterables of instances that should be created.
Source code in winidjango/src/commands/import_data.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
@abstractmethod
def get_bulks_by_model(
    self, df: pl.DataFrame
) -> dict[type[Model], Iterable[Model]]:
    """Map the cleaned DataFrame to model-instance bulks.

    The implementation should inspect the cleaned DataFrame and
    return a mapping where keys are Django model classes and values
    are iterables of unsaved model instances (or dataclass-like
    objects accepted by the project's bulk-creation utility).

    Args:
        df (pl.DataFrame): The cleaned and normalized DataFrame.

    Returns:
        dict[type[Model], Iterable[Model]]: Mapping from model classes
            to iterables of instances that should be created.
    """
get_cleaning_df_cls() abstractmethod

Return the CleaningDF subclass used to normalize the data.

The returned class will be instantiated with the raw DataFrame returned from :meth:handle_import and must provide the transformations required to prepare data for mapping into model instances.

Returns:

Type Description
type[CleaningDF]

type[CleaningDF]: A subclass of CleaningDF that performs the necessary normalization and validation.
Source code in winidjango/src/commands/import_data.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
@abstractmethod
def get_cleaning_df_cls(self) -> type[CleaningDF]:
    """Return the `CleaningDF` subclass used to normalize the data.

    The returned class will be instantiated with the raw DataFrame
    returned from :meth:`handle_import` and must provide the
    transformations required to prepare data for mapping into model
    instances.

    Returns:
        type[CleaningDF]: A subclass of ``CleaningDF`` that performs
            the necessary normalization and validation.
    """
get_option(option)

Retrieve a parsed command option by key.

Parameters:

Name Type Description Default
option str

The option key to retrieve from self.options.

 required

Returns:

Name Type Description
Any Any

The value for the requested option. If the option is not present a KeyError will be raised (matching how Django exposes options in management commands).
Source code in winidjango/src/commands/base/base.py 
173
174
175
176
177
178
179
180
181
182
183
184
def get_option(self, option: str) -> Any:
    """Retrieve a parsed command option by key.

    Args:
        option (str): The option key to retrieve from ``self.options``.

    Returns:
        Any: The value for the requested option. If the option is not
            present a ``KeyError`` will be raised (matching how Django
            exposes options in management commands).
    """
    return self.options[option]
handle(*args, **options)

Orchestrate command execution.

Performs shared pre-processing by calling :meth:base_handle and then delegates to :meth:handle_command which must be implemented by subclasses.

Parameters:

Name Type Description Default
*args Any

Positional arguments forwarded from Django.

 ()
**options Any

Parsed command-line options.

 {}
Source code in winidjango/src/commands/base/base.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
def handle(self, *args: Any, **options: Any) -> None:
    """Orchestrate command execution.

    Performs shared pre-processing by calling :meth:`base_handle` and
    then delegates to :meth:`handle_command` which must be implemented
    by subclasses.

    Args:
        *args: Positional arguments forwarded from Django.
        **options: Parsed command-line options.
    """
    self.base_handle(*args, **options)
    self.handle_command()
handle_command()

Execute the full import lifecycle.

This template method reads raw data via :meth:handle_import, wraps it with the cleaning class returned by :meth:get_cleaning_df_cls and then persists the resulting model bulks returned by :meth:get_bulks_by_model.

Source code in winidjango/src/commands/import_data.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def handle_command(self) -> None:
    """Execute the full import lifecycle.

    This template method reads raw data via :meth:`handle_import`,
    wraps it with the cleaning class returned by
    :meth:`get_cleaning_df_cls` and then persists the resulting
    model bulks returned by :meth:`get_bulks_by_model`.
    """
    data_df = self.handle_import()

    cleaning_df_cls = self.get_cleaning_df_cls()
    self.cleaning_df = cleaning_df_cls(data_df)

    self.import_to_db()
handle_import() abstractmethod

Read raw data from the import source.

This method should read data from whatever source the concrete command targets (files, remote APIs, etc.) and return it as a polars.DataFrame. No cleaning should be performed here; cleaning is handled by the cleaning CleaningDF returned from get_cleaning_df_cls.

Returns:

Type Description
DataFrame

pl.DataFrame: Raw (uncleaned) tabular data to be cleaned and mapped to model instances.
Source code in winidjango/src/commands/import_data.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
@abstractmethod
def handle_import(self) -> pl.DataFrame:
    """Read raw data from the import source.

    This method should read data from whatever source the concrete
    command targets (files, remote APIs, etc.) and return it as a
    ``polars.DataFrame``. No cleaning should be performed here;
    cleaning is handled by the cleaning `CleaningDF` returned from
    ``get_cleaning_df_cls``.

    Returns:
        pl.DataFrame: Raw (uncleaned) tabular data to be cleaned and
            mapped to model instances.
    """
import_to_db()

Persist prepared model bulks to the database.

Calls the project's bulk_create_bulks_in_steps helper with the mapping returned from :meth:get_bulks_by_model.

Source code in winidjango/src/commands/import_data.py 
109
110
111
112
113
114
115
116
117
def import_to_db(self) -> None:
    """Persist prepared model bulks to the database.

    Calls the project's `bulk_create_bulks_in_steps` helper with the
    mapping returned from :meth:`get_bulks_by_model`.
    """
    bulks_by_model = self.get_bulks_by_model(df=self.cleaning_df.df)

    bulk_create_bulks_in_steps(bulks_by_model)

db

Database utilities and common model helpers used in the project.

This package contains helpers for working with Django models, such as topological sorting of model classes according to foreign-key dependencies, a lightweight model hashing helper, and a project-wide BaseModel that adds common timestamp fields.

bulk

Utilities for performing bulk operations on Django models.

This module centralizes helpers used by importers and maintenance commands to create, update and delete large collections of model instances efficiently. It provides batching, optional concurrent execution, dependency-aware ordering and simulation helpers for previewing cascade deletions.

bulk_create_bulks_in_steps(bulk_by_class, step=STANDARD_BULK_SIZE)

Create multiple model-type bulks in dependency order.

The function topologically sorts the provided model classes so that models referenced by foreign keys are created before models that reference them. Each class' instances are created in batches using :func:bulk_create_in_steps.

Parameters:

Name Type Description Default
bulk_by_class dict[type[Model], Iterable[Model]]

Mapping from model class to iterable of instances to create.

 required
step int

Chunk size for each model's batched creation.

 STANDARD_BULK_SIZE

Returns:

Type Description
dict[type[Model], list[Model]]

Mapping from model class to list of created instances.
Source code in winidjango/src/db/bulk.py 
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
def bulk_create_bulks_in_steps(
    bulk_by_class: dict[type[Model], Iterable[Model]],
    step: int = STANDARD_BULK_SIZE,
) -> dict[type[Model], list[Model]]:
    """Create multiple model-type bulks in dependency order.

    The function topologically sorts the provided model classes so that
    models referenced by foreign keys are created before models that
    reference them. Each class' instances are created in batches using
    :func:`bulk_create_in_steps`.

    Args:
        bulk_by_class: Mapping from model class to iterable of instances to create.
        step: Chunk size for each model's batched creation.

    Returns:
        Mapping from model class to list of created instances.
    """
    # order the bulks in order of creation depending how they depend on each other
    models_ = list(bulk_by_class.keys())
    ordered_models = topological_sort_models(models=models_)

    results: dict[type[Model], list[Model]] = {}
    for model_ in ordered_models:
        bulk = bulk_by_class[model_]
        result = bulk_create_in_steps(model=model_, bulk=bulk, step=step)
        results[model_] = result

    return results
bulk_create_in_steps(model, bulk, step=STANDARD_BULK_SIZE)

Create objects in batches and return created instances.

Breaks bulk into chunks of size step and calls the project's batched bulk-create helper for each chunk. Execution is performed using the concurrent utility where configured for throughput.

Parameters:

Name Type Description Default
model type[TModel]

Django model class to create instances for.

 required
bulk Iterable[TModel]

Iterable of unsaved model instances.

 required
step int

Number of instances to create per chunk.

 STANDARD_BULK_SIZE

Returns:

Type Description
list[TModel]

List of created model instances (flattened across chunks).
Source code in winidjango/src/db/bulk.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def bulk_create_in_steps[TModel: Model](
    model: type[TModel],
    bulk: Iterable[TModel],
    step: int = STANDARD_BULK_SIZE,
) -> list[TModel]:
    """Create objects in batches and return created instances.

    Breaks ``bulk`` into chunks of size ``step`` and calls the project's
    batched bulk-create helper for each chunk. Execution is performed
    using the concurrent utility where configured for throughput.

    Args:
        model: Django model class to create instances for.
        bulk: Iterable of unsaved model instances.
        step: Number of instances to create per chunk.

    Returns:
        List of created model instances (flattened across chunks).
    """
    return cast(
        "list[TModel]",
        bulk_method_in_steps(model=model, bulk=bulk, step=step, mode=MODE_CREATE),
    )
bulk_delete(model, objs, **_)

Delete the provided objects and return Django's delete summary.

Accepts either a QuerySet or an iterable of model instances. When an iterable of instances is provided it is converted to a QuerySet by filtering on primary keys before calling QuerySet.delete().

Parameters:

Name Type Description Default
model type[Model]

Django model class.

 required
objs Iterable[Model]

Iterable of model instances or a QuerySet.

 required

Returns:

Type Description
tuple[int, dict[str, int]]

Tuple(total_deleted, per_model_counts) as returned by delete().
Source code in winidjango/src/db/bulk.py 
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
def bulk_delete(
    model: type[Model], objs: Iterable[Model], **_: Any
) -> tuple[int, dict[str, int]]:
    """Delete the provided objects and return Django's delete summary.

    Accepts either a QuerySet or an iterable of model instances. When an
    iterable of instances is provided it is converted to a QuerySet by
    filtering on primary keys before calling ``QuerySet.delete()``.

    Args:
        model: Django model class.
        objs: Iterable of model instances or a QuerySet.

    Returns:
        Tuple(total_deleted, per_model_counts) as returned by ``delete()``.
    """
    return model.objects.filter(pk__in=(obj.pk for obj in objs)).delete()
bulk_delete_in_steps(model, bulk, step=STANDARD_BULK_SIZE)

Delete objects in batches and return deletion statistics.

Each chunk is deleted using Django's QuerySet delete which returns a (count, per-model-counts) tuple. Results are aggregated across chunks and returned as a consolidated tuple.

Parameters:

Name Type Description Default
model type[TModel]

Django model class.

 required
bulk Iterable[TModel]

Iterable of model instances to delete.

 required
step int

Chunk size for deletions.

 STANDARD_BULK_SIZE

Returns:

Type Description
int

A tuple containing the total number of deleted objects and a
dict[str, int]

mapping from model label to deleted count (including cascaded
tuple[int, dict[str, int]]

deletions).
Source code in winidjango/src/db/bulk.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def bulk_delete_in_steps[TModel: Model](
    model: type[TModel], bulk: Iterable[TModel], step: int = STANDARD_BULK_SIZE
) -> tuple[int, dict[str, int]]:
    """Delete objects in batches and return deletion statistics.

    Each chunk is deleted using Django's QuerySet ``delete`` which
    returns a (count, per-model-counts) tuple. Results are aggregated
    across chunks and returned as a consolidated tuple.

    Args:
        model: Django model class.
        bulk: Iterable of model instances to delete.
        step: Chunk size for deletions.

    Returns:
        A tuple containing the total number of deleted objects and a
        mapping from model label to deleted count (including cascaded
        deletions).
    """
    return cast(
        "tuple[int, dict[str, int]]",
        bulk_method_in_steps(
            model=model,
            bulk=bulk,
            step=step,
            mode=MODE_DELETE,
        ),
    )
bulk_method_in_steps(model, bulk, step, mode, **kwargs) 
bulk_method_in_steps(
    model: type[TModel],
    bulk: Iterable[TModel],
    step: int,
    mode: Literal["create"],
    **kwargs: Any,
) -> list[TModel]

bulk_method_in_steps(
    model: type[TModel],
    bulk: Iterable[TModel],
    step: int,
    mode: Literal["update"],
    **kwargs: Any,
) -> int

bulk_method_in_steps(
    model: type[TModel],
    bulk: Iterable[TModel],
    step: int,
    mode: Literal["delete"],
    **kwargs: Any,
) -> tuple[int, dict[str, int]]

Run a batched bulk operation (create/update/delete) on bulk.

This wrapper warns if called from within an existing transaction and delegates actual work to :func:bulk_method_in_steps_atomic which is executed inside an atomic transaction. The return type depends on mode (see :mod:winidjango.src.db.bulk constants).

Parameters:

Name Type Description Default
model type[TModel]

Django model class to operate on.

 required
bulk Iterable[TModel]

Iterable of model instances.

 required
step int

Chunk size for processing.

 required
mode MODE_TYPES

One of 'create', 'update' or 'delete'.

 required
**kwargs Any

Additional keyword arguments forwarded to the underlying bulk methods (for example update_fields for updates).

 {}

Returns:

Type Description
int | tuple[int, dict[str, int]] | list[TModel]

For create: list of created instances.
int | tuple[int, dict[str, int]] | list[TModel]

For update: integer number of updated rows.
int | tuple[int, dict[str, int]] | list[TModel]

For delete: tuple(total_deleted, per_model_counts).
Source code in winidjango/src/db/bulk.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
def bulk_method_in_steps[TModel: Model](
    model: type[TModel],
    bulk: Iterable[TModel],
    step: int,
    mode: MODE_TYPES,
    **kwargs: Any,
) -> int | tuple[int, dict[str, int]] | list[TModel]:
    """Run a batched bulk operation (create/update/delete) on ``bulk``.

    This wrapper warns if called from within an existing transaction and
    delegates actual work to :func:`bulk_method_in_steps_atomic` which is
    executed inside an atomic transaction. The return type depends on
    ``mode`` (see :mod:`winidjango.src.db.bulk` constants).

    Args:
        model: Django model class to operate on.
        bulk: Iterable of model instances.
        step: Chunk size for processing.
        mode: One of ``'create'``, ``'update'`` or ``'delete'``.
        **kwargs: Additional keyword arguments forwarded to the underlying
            bulk methods (for example ``update_fields`` for updates).

    Returns:
        For ``create``: list of created instances.
        For ``update``: integer number of updated rows.
        For ``delete``: tuple(total_deleted, per_model_counts).
    """
    # check if we are inside a transaction.atomic block
    _in_atomic_block = transaction.get_connection().in_atomic_block
    if _in_atomic_block:
        logger.info(
            "BE CAREFUL USING BULK OPERATIONS INSIDE A BROADER TRANSACTION BLOCK. "
            "BULKING WITH BULKS THAT DEPEND ON EACH OTHER CAN CAUSE "
            "INTEGRITY ERRORS OR POTENTIAL OTHER ISSUES."
        )
    return bulk_method_in_steps_atomic(
        model=model, bulk=bulk, step=step, mode=mode, **kwargs
    )
bulk_method_in_steps_atomic(model, bulk, step, mode, **kwargs) 
bulk_method_in_steps_atomic(
    model: type[TModel],
    bulk: Iterable[TModel],
    step: int,
    mode: Literal["create"],
    **kwargs: Any,
) -> list[TModel]

bulk_method_in_steps_atomic(
    model: type[TModel],
    bulk: Iterable[TModel],
    step: int,
    mode: Literal["update"],
    **kwargs: Any,
) -> int

bulk_method_in_steps_atomic(
    model: type[TModel],
    bulk: Iterable[TModel],
    step: int,
    mode: Literal["delete"],
    **kwargs: Any,
) -> tuple[int, dict[str, int]]

Atomic implementation of the batched bulk operation.

This function is decorated with transaction.atomic so each call to this function runs in a database transaction. Note that nesting additional, outer atomic blocks that also include dependent bulk operations can cause integrity issues for operations that depend on each other's side-effects; callers should be careful about atomic decorator placement in higher-level code.

Parameters:

Name Type Description Default
model type[TModel]

Django model class.

 required
bulk Iterable[TModel]

Iterable of model instances.

 required
step int

Chunk size for processing.

 required
mode MODE_TYPES

One of 'create', 'update' or 'delete'.

 required
**kwargs Any

Forwarded to the underlying bulk method.

 {}

Returns:

Name Type Description
See int | tuple[int, dict[str, int]] | list[TModel]

func:bulk_method_in_steps for return value semantics.
Source code in winidjango/src/db/bulk.py 
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
@transaction.atomic
def bulk_method_in_steps_atomic[TModel: Model](
    model: type[TModel],
    bulk: Iterable[TModel],
    step: int,
    mode: MODE_TYPES,
    **kwargs: Any,
) -> int | tuple[int, dict[str, int]] | list[TModel]:
    """Atomic implementation of the batched bulk operation.

    This function is decorated with ``transaction.atomic`` so each call
    to this function runs in a database transaction. Note that nesting
    additional, outer atomic blocks that also include dependent bulk
    operations can cause integrity issues for operations that depend on
    each other's side-effects; callers should be careful about atomic
    decorator placement in higher-level code.

    Args:
        model: Django model class.
        bulk: Iterable of model instances.
        step: Chunk size for processing.
        mode: One of ``'create'``, ``'update'`` or ``'delete'``.
        **kwargs: Forwarded to the underlying bulk method.

    Returns:
        See :func:`bulk_method_in_steps` for return value semantics.
    """
    bulk_method = get_bulk_method(model=model, mode=mode, **kwargs)

    chunks = get_step_chunks(bulk=bulk, step=step)

    # multithreading significantly increases speed
    result = multithread_loop(
        process_function=bulk_method,
        process_args=chunks,
    )

    return flatten_bulk_in_steps_result(result=result, mode=mode)
bulk_update_in_steps(model, bulk, update_fields, step=STANDARD_BULK_SIZE)

Update objects in batches and return total updated count.

Parameters:

Name Type Description Default
model type[TModel]

Django model class.

 required
bulk Iterable[TModel]

Iterable of model instances to update (must have PKs set).

 required
update_fields list[str]

Fields to update on each instance when calling bulk_update.

 required
step int

Chunk size for batched updates.

 STANDARD_BULK_SIZE

Returns:

Type Description
int

Total number of rows updated across all chunks.
Source code in winidjango/src/db/bulk.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def bulk_update_in_steps[TModel: Model](
    model: type[TModel],
    bulk: Iterable[TModel],
    update_fields: list[str],
    step: int = STANDARD_BULK_SIZE,
) -> int:
    """Update objects in batches and return total updated count.

    Args:
        model: Django model class.
        bulk: Iterable of model instances to update (must have PKs set).
        update_fields: Fields to update on each instance when calling
            ``bulk_update``.
        step: Chunk size for batched updates.

    Returns:
        Total number of rows updated across all chunks.
    """
    return cast(
        "int",
        bulk_method_in_steps(
            model=model, bulk=bulk, step=step, mode=MODE_UPDATE, fields=update_fields
        ),
    )
flatten_bulk_in_steps_result(result, mode) 
flatten_bulk_in_steps_result(
    result: list[list[TModel]], mode: Literal["create"]
) -> list[TModel]

flatten_bulk_in_steps_result(
    result: list[int], mode: Literal["update"]
) -> int

flatten_bulk_in_steps_result(
    result: list[tuple[int, dict[str, int]]],
    mode: Literal["delete"],
) -> tuple[int, dict[str, int]]

Aggregate per-chunk results returned by concurrent bulk execution.

Depending on mode the function reduces a list of per-chunk results into a single consolidated return value:

  • create: flattens a list of lists into a single list of objects
  • update: sums integer counts returned per chunk
  • delete: aggregates (count, per-model-dict) tuples into a single total and combined per-model counts

Parameters:

Name Type Description Default
result list[int] | list[tuple[int, dict[str, int]]] | list[list[TModel]]

List of per-chunk results returned by the chunk function.

 required
mode str

One of the supported modes.

 required

Returns:

Type Description
int | tuple[int, dict[str, int]] | list[TModel]

Aggregated result corresponding to mode.
Source code in winidjango/src/db/bulk.py 
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
def flatten_bulk_in_steps_result[TModel: Model](
    result: list[int] | list[tuple[int, dict[str, int]]] | list[list[TModel]], mode: str
) -> int | tuple[int, dict[str, int]] | list[TModel]:
    """Aggregate per-chunk results returned by concurrent bulk execution.

    Depending on ``mode`` the function reduces a list of per-chunk
    results into a single consolidated return value:

    - ``create``: flattens a list of lists into a single list of objects
    - ``update``: sums integer counts returned per chunk
    - ``delete``: aggregates (count, per-model-dict) tuples into a single
      total and combined per-model counts

    Args:
        result: List of per-chunk results returned by the chunk function.
        mode: One of the supported modes.

    Returns:
        Aggregated result corresponding to ``mode``.
    """
    if mode == MODE_UPDATE:
        # formated as [1000, 1000, ...]
        # since django 4.2 bulk_update returns the count of updated objects
        result = cast("list[int]", result)
        return int(sum(result))
    if mode == MODE_DELETE:
        # formated as [(count, {model_name: count, model_cascade_name: count}), ...]
        # join the results to get the total count of deleted objects
        result = cast("list[tuple[int, dict[str, int]]]", result)
        total_count = 0
        count_sum_by_model: defaultdict[str, int] = defaultdict(int)
        for count_sum, count_by_model in result:
            total_count += count_sum
            for model_name, count in count_by_model.items():
                count_sum_by_model[model_name] += count
        return (total_count, dict(count_sum_by_model))
    if mode == MODE_CREATE:
        # formated as [[obj1, obj2, ...], [obj1, obj2, ...], ...]
        result = cast("list[list[TModel]]", result)
        return [item for sublist in result for item in sublist]

    msg = f"Invalid method. Must be one of {MODES}"
    raise ValueError(msg)
get_bulk_method(model, mode, **kwargs) 
get_bulk_method(
    model: type[Model],
    mode: Literal["create"],
    **kwargs: Any,
) -> Callable[[list[Model]], list[Model]]

get_bulk_method(
    model: type[Model],
    mode: Literal["update"],
    **kwargs: Any,
) -> Callable[[list[Model]], int]

get_bulk_method(
    model: type[Model],
    mode: Literal["delete"],
    **kwargs: Any,
) -> Callable[[list[Model]], tuple[int, dict[str, int]]]

Return a callable that performs the requested bulk operation on a chunk.

The returned function accepts a single argument (a list of model instances) and returns the per-chunk result for the chosen mode.

Parameters:

Name Type Description Default
model type[Model]

Django model class.

 required
mode MODE_TYPES

One of 'create', 'update' or 'delete'.

 required
**kwargs Any

Forwarded to the underlying ORM bulk methods.

 {}

Raises:

Type Description
ValueError

If mode is invalid.

Returns:

Type Description
Callable[[list[Model]], list[Model] | int | tuple[int, dict[str, int]]]

Callable that accepts a list of model instances and returns the
Callable[[list[Model]], list[Model] | int | tuple[int, dict[str, int]]]

result for that chunk.
Source code in winidjango/src/db/bulk.py 
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
def get_bulk_method(
    model: type[Model], mode: MODE_TYPES, **kwargs: Any
) -> Callable[[list[Model]], list[Model] | int | tuple[int, dict[str, int]]]:
    """Return a callable that performs the requested bulk operation on a chunk.

    The returned function accepts a single argument (a list of model
    instances) and returns the per-chunk result for the chosen mode.

    Args:
        model: Django model class.
        mode: One of ``'create'``, ``'update'`` or ``'delete'``.
        **kwargs: Forwarded to the underlying ORM bulk methods.

    Raises:
        ValueError: If ``mode`` is invalid.

    Returns:
        Callable that accepts a list of model instances and returns the
        result for that chunk.
    """
    bulk_method: Callable[[list[Model]], list[Model] | int | tuple[int, dict[str, int]]]
    if mode == MODE_CREATE:

        def bulk_create_chunk(chunk: list[Model]) -> list[Model]:
            return model.objects.bulk_create(objs=chunk, **kwargs)

        bulk_method = bulk_create_chunk
    elif mode == MODE_UPDATE:

        def bulk_update_chunk(chunk: list[Model]) -> int:
            return model.objects.bulk_update(objs=chunk, **kwargs)

        bulk_method = bulk_update_chunk
    elif mode == MODE_DELETE:

        def bulk_delete_chunk(chunk: list[Model]) -> tuple[int, dict[str, int]]:
            return bulk_delete(model=model, objs=chunk, **kwargs)

        bulk_method = bulk_delete_chunk
    else:
        msg = f"Invalid method. Must be one of {MODES}"
        raise ValueError(msg)

    return bulk_method
get_differences_between_bulks(bulk1, bulk2, fields)

Return differences and intersections between two bulks of the same model.

Instances are compared using :func:hash_model_instance over the provided fields. The function maintains the original ordering for returned lists so that callers can preserve deterministic ordering when applying diffs.

Parameters:

Name Type Description Default
bulk1 list[TModel1]

First list of model instances.

 required
bulk2 list[TModel2]

Second list of model instances.

 required
fields list[Field[Any, Any] | ForeignObjectRel | GenericForeignKey]

Fields to include when hashing instances.

 required

Raises:

Type Description
ValueError

If bulks are empty or contain different model types.

Returns:

Type Description
tuple[list[TModel1], list[TModel2], list[TModel1], list[TModel2]]

Four lists in the order: (in_1_not_2, in_2_not_1, in_both_from_1, in_both_from_2).
Source code in winidjango/src/db/bulk.py 
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
def get_differences_between_bulks[TModel1: Model, TModel2: Model](
    bulk1: list[TModel1],
    bulk2: list[TModel2],
    fields: "list[Field[Any, Any] | ForeignObjectRel | GenericForeignKey]",
) -> tuple[list[TModel1], list[TModel2], list[TModel1], list[TModel2]]:
    """Return differences and intersections between two bulks of the same model.

    Instances are compared using :func:`hash_model_instance` over the
    provided ``fields``. The function maintains the original ordering
    for returned lists so that callers can preserve deterministic
    ordering when applying diffs.

    Args:
        bulk1: First list of model instances.
        bulk2: Second list of model instances.
        fields: Fields to include when hashing instances.

    Raises:
        ValueError: If bulks are empty or contain different model types.

    Returns:
        Four lists in the order:
            (in_1_not_2, in_2_not_1, in_both_from_1, in_both_from_2).
    """
    if not bulk1 or not bulk2:
        return bulk1, bulk2, [], []

    if type(bulk1[0]) is not type(bulk2[0]):
        msg = "Both bulks must be of the same model type."
        raise ValueError(msg)

    hash_model_instance_with_fields = partial(
        hash_model_instance,
        fields=fields,
    )
    # Precompute hashes and map them directly to models in a single pass for both bulks
    hashes1 = list(map(hash_model_instance_with_fields, bulk1))
    hashes2 = list(map(hash_model_instance_with_fields, bulk2))

    # Convert keys to sets for difference operations
    set1, set2 = set(hashes1), set(hashes2)

    # Calculate differences between sets
    # Find differences and intersection with original order preserved
    # Important, we need to return the original objects that are the same in memory,
    # so in_1_not_2 and in_2_not_1
    in_1_not_2 = set1 - set2
    in_1_not_2_list = [
        model
        for model, hash_ in zip(bulk1, hashes1, strict=False)
        if hash_ in in_1_not_2
    ]

    in_2_not_1 = set2 - set1
    in_2_not_1_list = [
        model
        for model, hash_ in zip(bulk2, hashes2, strict=False)
        if hash_ in in_2_not_1
    ]

    in_1_and_2 = set1 & set2
    in_1_and_2_from_1 = [
        model
        for model, hash_ in zip(bulk1, hashes1, strict=False)
        if hash_ in in_1_and_2
    ]
    in_1_and_2_from_2 = [
        model
        for model, hash_ in zip(bulk2, hashes2, strict=False)
        if hash_ in in_1_and_2
    ]

    return in_1_not_2_list, in_2_not_1_list, in_1_and_2_from_1, in_1_and_2_from_2
get_step_chunks(bulk, step)

Yield consecutive chunks of at most step items from bulk.

The function yields a single-tuple containing the chunk (a list of model instances) because the concurrent execution helper expects a tuple of positional arguments for the target function.

Parameters:

Name Type Description Default
bulk Iterable[Model]

Iterable of model instances.

 required
step int

Maximum number of instances per yielded chunk.

 required

Yields:

Type Description
tuple[list[Model]]

Tuples where the first element is a list of model instances.
Source code in winidjango/src/db/bulk.py 
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
def get_step_chunks(
    bulk: Iterable[Model], step: int
) -> Generator[tuple[list[Model]], None, None]:
    """Yield consecutive chunks of at most ``step`` items from ``bulk``.

    The function yields a single-tuple containing the chunk (a list of
    model instances) because the concurrent execution helper expects a
    tuple of positional arguments for the target function.

    Args:
        bulk: Iterable of model instances.
        step: Maximum number of instances per yielded chunk.

    Yields:
        Tuples where the first element is a list of model instances.
    """
    bulk = iter(bulk)
    while True:
        chunk = list(islice(bulk, step))
        if not chunk:
            break
        yield (chunk,)  # bc concurrent_loop expects a tuple of args
multi_simulate_bulk_deletion(entries)

Simulate deletions for multiple model classes and merge results.

Runs :func:simulate_bulk_deletion for each provided model and returns a unified mapping of all models that would be deleted.

Parameters:

Name Type Description Default
entries dict[type[Model], list[Model]]

Mapping from model class to list of instances to simulate.

 required

Returns:

Type Description
dict[type[Model], set[Model]]

Mapping from model class to set of instances that would be deleted.
Source code in winidjango/src/db/bulk.py 
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
def multi_simulate_bulk_deletion(
    entries: dict[type[Model], list[Model]],
) -> dict[type[Model], set[Model]]:
    """Simulate deletions for multiple model classes and merge results.

    Runs :func:`simulate_bulk_deletion` for each provided model and
    returns a unified mapping of all models that would be deleted.

    Args:
        entries: Mapping from model class to list of instances to simulate.

    Returns:
        Mapping from model class to set of instances that would be deleted.
    """
    deletion_summaries = [
        simulate_bulk_deletion(model, entry) for model, entry in entries.items()
    ]
    # join the dicts to get the total count of deleted objects
    joined_deletion_summary: defaultdict[type[Model], set[Model]] = defaultdict(set)
    for deletion_summary in deletion_summaries:
        for model, objects in deletion_summary.items():
            joined_deletion_summary[model].update(objects)

    return dict(joined_deletion_summary)
simulate_bulk_deletion(model_class, entries)

Simulate Django's delete cascade and return affected objects.

Uses :class:django.db.models.deletion.Collector to determine which objects (including cascaded related objects) would be removed if the provided entries were deleted. No database writes are performed.

Parameters:

Name Type Description Default
model_class type[TModel]

Model class of the provided entries.

 required
entries list[TModel]

Instances to simulate deletion for.

 required

Returns:

Type Description
dict[type[Model], set[Model]]

Mapping from model class to set of instances that would be deleted.
Source code in winidjango/src/db/bulk.py 
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
def simulate_bulk_deletion[TModel: Model](
    model_class: type[TModel], entries: list[TModel]
) -> dict[type[Model], set[Model]]:
    """Simulate Django's delete cascade and return affected objects.

    Uses :class:`django.db.models.deletion.Collector` to determine which
    objects (including cascaded related objects) would be removed if the
    provided entries were deleted. No database writes are performed.

    Args:
        model_class: Model class of the provided entries.
        entries: Instances to simulate deletion for.

    Returns:
        Mapping from model class to set of instances that would be deleted.
    """
    if not entries:
        return {}

    # Initialize the Collector
    using = router.db_for_write(model_class)
    collector = Collector(using)

    # Collect deletion cascade for all entries
    collector.collect(entries)  # ty:ignore[invalid-argument-type]

    # Prepare the result dictionary
    deletion_summary: defaultdict[type[Model], set[Model]] = defaultdict(set)

    # Add normal deletes
    for model, objects in collector.data.items():
        deletion_summary[model].update(objects)  # objects is already iterable

    # Add fast deletes (explicitly expand querysets)
    for queryset in collector.fast_deletes:
        deletion_summary[queryset.model].update(list(queryset))

    return deletion_summary

fields

Utilities for inspecting Django model fields.

This module provides small helpers that make it easier to introspect Django model fields and metadata in a type-friendly way. The helpers are used across the project's database utilities to implement operations like topological sorting and deterministic hashing of model instances.

get_field_names(fields)

Return the name attribute for a list of Django field objects.

Parameters:

Name Type Description Default
fields list[Field | ForeignObjectRel | GenericForeignKey]

Field objects obtained from a model's _meta.get_fields().

 required

Returns:

Type Description
list[str]

list[str]: List of field names in the same order as fields.
Source code in winidjango/src/db/fields.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
def get_field_names(
    fields: "list[Field[Any, Any] | ForeignObjectRel | GenericForeignKey]",
) -> list[str]:
    """Return the ``name`` attribute for a list of Django field objects.

    Args:
        fields (list[Field | ForeignObjectRel | GenericForeignKey]):
            Field objects obtained from a model's ``_meta.get_fields()``.

    Returns:
        list[str]: List of field names in the same order as ``fields``.
    """
    return [field.name for field in fields]
get_fields(model)

Return all field objects for a Django model.

This wraps model._meta.get_fields() and is typed to include relationship fields so callers can handle both regular and related fields uniformly.

Parameters:

Name Type Description Default
model type[Model]

Django model class.

 required

Returns:

Type Description
list[Field[Any, Any] | ForeignObjectRel | GenericForeignKey]

list[Field | ForeignObjectRel | GenericForeignKey]: All field objects associated with the model.
Source code in winidjango/src/db/fields.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def get_fields[TModel: Model](
    model: type[TModel],
) -> "list[Field[Any, Any] | ForeignObjectRel | GenericForeignKey]":
    """Return all field objects for a Django model.

    This wraps ``model._meta.get_fields()`` and is typed to include
    relationship fields so callers can handle both regular and related
    fields uniformly.

    Args:
        model (type[Model]): Django model class.

    Returns:
        list[Field | ForeignObjectRel | GenericForeignKey]: All field
            objects associated with the model.
    """
    return get_model_meta(model).get_fields()
get_model_meta(model)

Return a model class' _meta options object.

This small wrapper exists to make typing clearer at call sites where the code needs the model Options object.

Parameters:

Name Type Description Default
model type[Model]

Django model class.

 required

Returns:

Name Type Description
Options Options[Model]

The model's _meta options object.
Source code in winidjango/src/db/fields.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
def get_model_meta(model: type[Model]) -> "Options[Model]":
    """Return a model class' ``_meta`` options object.

    This small wrapper exists to make typing clearer at call sites where
    the code needs the model Options object.

    Args:
        model (type[Model]): Django model class.

    Returns:
        Options: The model's ``_meta`` options object.
    """
    return model._meta  # noqa: SLF001

models

Database utilities and lightweight model helpers.

This module provides helpers used across the project when manipulating Django models: ordering models by foreign-key dependencies, creating a deterministic hash for unsaved instances, and a project-wide BaseModel that exposes common timestamp fields.

BaseModel

Bases: Model

Abstract base model containing common fields and helpers.

Concrete models can inherit from this class to get consistent created_at and updated_at timestamp fields and convenient string representations.

Source code in winidjango/src/db/models.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
class BaseModel(Model):
    """Abstract base model containing common fields and helpers.

    Concrete models can inherit from this class to get consistent
    ``created_at`` and ``updated_at`` timestamp fields and convenient
    string representations.
    """

    created_at: DateTimeField[datetime, datetime] = DateTimeField(auto_now_add=True)
    updated_at: DateTimeField[datetime, datetime] = DateTimeField(auto_now=True)

    class Meta:
        """Mark the model as abstract."""

        # abstract does not inherit in children
        abstract = True

    def __str__(self) -> str:
        """Return a concise human-readable representation.

        The default shows the model class name and primary key which is
        useful for logging and interactive debugging.

        Returns:
            str: Short representation, e.g. ``MyModel(123)``.
        """
        return f"{self.__class__.__name__}({self.pk})"

    def __repr__(self) -> str:
        """Base representation of a model."""
        return str(self)

    @property
    def meta(self) -> "Options[Self]":
        """Return the model's ``_meta`` options object.

        This property is a small convenience wrapper used to make access
        sites slightly more explicit in code and improve typing in callers.
        """
        return self._meta
meta property

Return the model's _meta options object.

This property is a small convenience wrapper used to make access sites slightly more explicit in code and improve typing in callers.

Meta

Mark the model as abstract.

Source code in winidjango/src/db/models.py 
130
131
132
133
134
class Meta:
    """Mark the model as abstract."""

    # abstract does not inherit in children
    abstract = True
__repr__()

Base representation of a model.

Source code in winidjango/src/db/models.py 
147
148
149
def __repr__(self) -> str:
    """Base representation of a model."""
    return str(self)
__str__()

Return a concise human-readable representation.

The default shows the model class name and primary key which is useful for logging and interactive debugging.

Returns:

Name Type Description
str str

Short representation, e.g. MyModel(123).
Source code in winidjango/src/db/models.py 
136
137
138
139
140
141
142
143
144
145
def __str__(self) -> str:
    """Return a concise human-readable representation.

    The default shows the model class name and primary key which is
    useful for logging and interactive debugging.

    Returns:
        str: Short representation, e.g. ``MyModel(123)``.
    """
    return f"{self.__class__.__name__}({self.pk})"
hash_model_instance(instance, fields)

Compute a deterministic hash for a model instance.

The function returns a hash suitable for comparing unsaved model instances by their field values. If the instance has a primary key (instance.pk) that key is hashed and returned immediately; this keeps comparisons cheap for persisted objects.

Parameters:

Name Type Description Default
instance Model

The Django model instance to hash.

 required
fields list[Field | ForeignObjectRel | GenericForeignKey]

Field objects that should be included when computing the hash.

 required

Returns:

Name Type Description
int int

Deterministic integer hash of the instance. For persisted instances this is hash(instance.pk).
Notes
  • The returned hash is intended for heuristic comparisons (e.g. deduplication in import pipelines) and is not cryptographically secure. Use with care when relying on absolute uniqueness.
Source code in winidjango/src/db/models.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def hash_model_instance(
    instance: Model,
    fields: "list[Field[Any, Any] | ForeignObjectRel | GenericForeignKey]",
) -> int:
    """Compute a deterministic hash for a model instance.

    The function returns a hash suitable for comparing unsaved model
    instances by their field values. If the instance has a primary key
    (``instance.pk``) that key is hashed and returned immediately; this
    keeps comparisons cheap for persisted objects.

    Args:
        instance (Model): The Django model instance to hash.
        fields (list[Field | ForeignObjectRel | GenericForeignKey]):
            Field objects that should be included when computing the hash.

    Returns:
        int: Deterministic integer hash of the instance. For persisted
            instances this is ``hash(instance.pk)``.

    Notes:
        - The returned hash is intended for heuristic comparisons (e.g.
          deduplication in import pipelines) and is not cryptographically
          secure. Use with care when relying on absolute uniqueness.
    """
    if instance.pk:
        return hash(instance.pk)

    field_names = get_field_names(fields)
    model_dict = model_to_dict(instance, fields=field_names)
    sorted_dict = dict(sorted(model_dict.items()))
    values = (type(instance), tuple(sorted_dict.items()))
    return hash(values)
topological_sort_models(models)

Sort Django models in dependency order using topological sorting.

Analyzes foreign key relationships between Django models and returns them in an order where dependencies come before dependents. This ensures that when performing operations like bulk creation or deletion, models are processed in the correct order to avoid foreign key constraint violations.

The function uses Python's graphlib.TopologicalSorter to perform the sorting based on ForeignKey relationships between the provided models. Only relationships between models in the input list are considered.

Parameters:

Name Type Description Default
models list[type[Model]]

A list of Django model classes to sort based on their foreign key dependencies.

 required

Returns:

Type Description
list[type[Model]]

list[type[Model]]: The input models sorted in dependency order, where models that are referenced by foreign keys appear before models that reference them. Self-referential relationships are ignored.

Raises:

Type Description
CycleError

If there are circular dependencies between models that cannot be resolved.
Example
Assuming Author model has no dependencies
and Book model has ForeignKey to Author

models = [Book, Author] sorted_models = topological_sort_models(models) sorted_models [, ]

Note
  • Only considers ForeignKey relationships, not other field types
  • Self-referential foreign keys are ignored to avoid self-loops
  • Only relationships between models in the input list are considered
Source code in winidjango/src/db/models.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def topological_sort_models(
    models: list[type[Model]],
) -> list[type[Model]]:
    """Sort Django models in dependency order using topological sorting.

    Analyzes foreign key relationships between Django models and returns them
    in an order where dependencies come before dependents. This ensures that
    when performing operations like bulk creation or deletion, models are
    processed in the correct order to avoid foreign key constraint violations.

    The function uses Python's graphlib.TopologicalSorter to perform the sorting
    based on ForeignKey relationships between the provided models. Only
    relationships between models in the input list are considered.

    Args:
        models (list[type[Model]]): A list of Django model classes to sort
            based on their foreign key dependencies.

    Returns:
        list[type[Model]]: The input models sorted in dependency order, where
            models that are referenced by foreign keys appear before models
            that reference them. Self-referential relationships are ignored.

    Raises:
        graphlib.CycleError: If there are circular dependencies between models
            that cannot be resolved.

    Example:
        >>> # Assuming Author model has no dependencies
        >>> # and Book model has ForeignKey to Author
        >>> models = [Book, Author]
        >>> sorted_models = topological_sort_models(models)
        >>> sorted_models
        [<class 'Author'>, <class 'Book'>]

    Note:
        - Only considers ForeignKey relationships, not other field types
        - Self-referential foreign keys are ignored to avoid self-loops
        - Only relationships between models in the input list are considered
    """
    ts: TopologicalSorter[type[Model]] = TopologicalSorter()

    for model in models:
        deps = {
            field.related_model
            for field in get_fields(model)
            if isinstance(field, ForeignKey)
            and isinstance(field.related_model, type)
            and field.related_model in models
            and field.related_model is not model
        }
        ts.add(model, *deps)

    return list(ts.static_order())

sql

Low-level helper to execute raw SQL against Django's database.

This module exposes :func:execute_sql which runs a parameterized SQL query using Django's database connection and returns column names and rows. It is intended for one-off queries where ORM abstractions are insufficient or when reading complex reports from the database.

The helper uses Django's connection cursor context manager to ensure resources are cleaned up correctly. Results are fetched into memory so avoid using it for very large result sets.

execute_sql(sql, params=None)

Execute a SQL statement and return column names and rows.

Parameters:

Name Type Description Default
sql str

SQL statement possibly containing named placeholders (%(name)s) for database binding.

 required
params dict[str, Any] | None

Optional mapping of parameters to bind to the query.

 None

Returns:

Type Description
list[str]

Tuple[List[str], List[Tuple[Any, ...]]]: A tuple where the first
list[tuple[Any, ...]]

element is the list of column names (empty list if the statement
tuple[list[str], list[tuple[Any, ...]]]

returned no rows) and the second element is a list of row tuples.

Raises:

Type Description
Error

Propagates underlying database errors raised by Django's database backend.
Source code in winidjango/src/db/sql.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def execute_sql(
    sql: str, params: dict[str, Any] | None = None
) -> tuple[list[str], list[tuple[Any, ...]]]:
    """Execute a SQL statement and return column names and rows.

    Args:
        sql (str): SQL statement possibly containing named placeholders
            (``%(name)s``) for database binding.
        params (dict[str, Any] | None): Optional mapping of parameters to
            bind to the query.

    Returns:
        Tuple[List[str], List[Tuple[Any, ...]]]: A tuple where the first
        element is the list of column names (empty list if the statement
        returned no rows) and the second element is a list of row tuples.

    Raises:
        django.db.Error: Propagates underlying database errors raised by
            Django's database backend.
    """
    with connection.cursor() as cursor:
        cursor.execute(sql=sql, params=params)
        rows = cursor.fetchall()
        column_names = (
            [col[0] for col in cursor.description] if cursor.description else []
        )

    return column_names, rows

winidjango

init module.

rig

init module.

resources

init module.

tools

Tool wrappers for CLI tools used in development workflows.

Tools are subclasses of Tool providing methods that return Args objects for type-safe command construction and execution.

tools

Override pyrig tools.

ProjectTester

Bases: ProjectTester

Subclass of ProjectTester for customizing pyrig behavior.

Source code in winidjango/rig/tools/tools.py 
15
16
17
18
19
20
class ProjectTester(BaseProjectTester):
    """Subclass of ProjectTester for customizing pyrig behavior."""

    def dev_dependencies(self) -> tuple[str, ...]:
        """Get the dev dependencies."""
        return (*super().dev_dependencies(), "pytest-django")
dev_dependencies()

Get the dev dependencies.

Source code in winidjango/rig/tools/tools.py 
18
19
20
def dev_dependencies(self) -> tuple[str, ...]:
    """Get the dev dependencies."""
    return (*super().dev_dependencies(), "pytest-django")
Pyrigger

Bases: Pyrigger

Subclass of Pyrigger for customizing pyrig behavior.

Source code in winidjango/rig/tools/tools.py 
 7
 8
 9
10
11
12
class Pyrigger(BasePyrigger):
    """Subclass of Pyrigger for customizing pyrig behavior."""

    def dev_dependencies(self) -> tuple[str, ...]:
        """Get the dev dependencies."""
        return (*super().dev_dependencies(), "django-stubs")
dev_dependencies()

Get the dev dependencies.

Source code in winidjango/rig/tools/tools.py 
10
11
12
def dev_dependencies(self) -> tuple[str, ...]:
    """Get the dev dependencies."""
    return (*super().dev_dependencies(), "django-stubs")

src

src package.

This package exposes the project's internal modules used by the command-line utilities and database helpers. It exists primarily so that code under winidjango/src can be imported using the winidjango.src package path in other modules and tests.

The package itself contains the following subpackages: - commands - management command helpers and base classes - db - database utilities and model helpers

Consumers should import the specific submodules they need rather than relying on side effects from this package's import-time execution.

commands

Utilities and base classes for management commands.

The commands package contains base command classes and helpers used to implement Django management commands in the project. Subpackages and modules under commands provide reusable patterns for argument handling, logging and common command behaviors so that individual commands can focus on their business logic.

base

Base helpers and abstractions for management commands.

This package provides a common abstract base class used by project management commands. The base class centralizes argument handling, standard options (dry-run, batching, timeouts, etc.) and integrates logging behavior used throughout the commands package.

base

Utilities and an abstract base class for Django management commands.

This module defines :class:ABCBaseCommand, a reusable abstract base that combines Django's BaseCommand with the project's logging mixins and standard argument handling. The base class implements a template method pattern so concrete commands only need to implement the abstract extension points for providing command-specific arguments and business logic.

ABCBaseCommand

Bases: ABCLoggingMixin, BaseCommand

Abstract base class for management commands with logging and standard options.

The class wires common behavior such as base arguments (dry-run, batching, timeouts) and provides extension points that concrete commands must implement: :meth:add_command_arguments and :meth:handle_command.

Notes
  • Inheritance order matters: the logging mixin must precede BaseCommand so mixin initialization occurs as expected.
  • The base class follows the template method pattern; concrete commands should not override :meth:add_arguments or :meth:handle but implement the abstract hooks instead.
Source code in winidjango/src/commands/base/base.py 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
class ABCBaseCommand(ABCLoggingMixin, BaseCommand):
    """Abstract base class for management commands with logging and standard options.

    The class wires common behavior such as base arguments (dry-run,
    batching, timeouts) and provides extension points that concrete
    commands must implement: :meth:`add_command_arguments` and
    :meth:`handle_command`.

    Notes:
        - Inheritance order matters: the logging mixin must precede
          ``BaseCommand`` so mixin initialization occurs as expected.
        - The base class follows the template method pattern; concrete
          commands should not override :meth:`add_arguments` or
          :meth:`handle` but implement the abstract hooks instead.
    """

    class Options:
        """Just a container class for hard coding the option keys."""

        DRY_RUN = "dry_run"
        FORCE = "force"
        DELETE = "delete"
        YES = "yes"
        TIMEOUT = "timeout"
        BATCH_SIZE = "batch_size"
        THREADS = "threads"
        PROCESSES = "processes"

    def add_arguments(self, parser: ArgumentParser) -> None:
        """Configure command-line arguments for the command.

        Adds common base arguments (dry-run, force, delete, timeout,
        batching and concurrency options) and then delegates to
        :meth:`add_command_arguments` for command-specific options.

        Args:
            parser (ArgumentParser): The argument parser passed by Django.
        """
        # add base args that are used in most commands
        self.base_add_arguments(parser)

        # add additional args that are specific to the command
        self.add_command_arguments(parser)

    def base_add_arguments(self, parser: ArgumentParser) -> None:
        """Add the project's standard command-line arguments to ``parser``.

        Args:
            parser (ArgumentParser): The argument parser passed by Django.
        """
        parser.add_argument(
            f"--{self.Options.DRY_RUN}",
            action="store_true",
            help="Show what would be done without actually executing the changes",
        )

        parser.add_argument(
            f"--{self.Options.FORCE}",
            action="store_true",
            help="Force an action in a command",
        )

        parser.add_argument(
            f"--{self.Options.DELETE}",
            action="store_true",
            help="Deleting smth in a command",
        )

        parser.add_argument(
            f"--{self.Options.YES}",
            action="store_true",
            help="Answer yes to all prompts",
            default=False,
        )

        parser.add_argument(
            f"--{self.Options.TIMEOUT}",
            type=int,
            help="Timeout for a command",
            default=None,
        )

        parser.add_argument(
            f"--{self.Options.BATCH_SIZE}",
            type=int,
            default=None,
            help="Number of items to process in each batch",
        )

        parser.add_argument(
            f"--{self.Options.THREADS}",
            type=int,
            default=None,
            help="Number of threads to use for processing",
        )

        parser.add_argument(
            f"--{self.Options.PROCESSES}",
            type=int,
            default=None,
            help="Number of processes to use for processing",
        )

    @abstractmethod
    def add_command_arguments(self, parser: ArgumentParser) -> None:
        """Define command-specific arguments.

        Implement this hook to add options and positional arguments that are
        specific to the concrete management command.

        Args:
            parser (ArgumentParser): The argument parser passed by Django.
        """

    def handle(self, *args: Any, **options: Any) -> None:
        """Orchestrate command execution.

        Performs shared pre-processing by calling :meth:`base_handle` and
        then delegates to :meth:`handle_command` which must be implemented
        by subclasses.

        Args:
            *args: Positional arguments forwarded from Django.
            **options: Parsed command-line options.
        """
        self.base_handle(*args, **options)
        self.handle_command()

    def base_handle(self, *args: Any, **options: Any) -> None:
        """Perform common pre-processing for commands.

        Stores the incoming arguments and options on the instance for use
        by :meth:`handle_command` and subclasses.

        Args:
            *args: Positional arguments forwarded from Django.
            **options: Parsed command-line options.
        """
        self.args = args
        self.options = options

    @abstractmethod
    def handle_command(self) -> None:
        """Run the command-specific behavior.

        This abstract hook should be implemented by concrete commands to
        perform the command's main work. Implementations should read
        ``self.args`` and ``self.options`` which were set in
        :meth:`base_handle`.
        """

    def get_option(self, option: str) -> Any:
        """Retrieve a parsed command option by key.

        Args:
            option (str): The option key to retrieve from ``self.options``.

        Returns:
            Any: The value for the requested option. If the option is not
                present a ``KeyError`` will be raised (matching how Django
                exposes options in management commands).
        """
        return self.options[option]
Options

Just a container class for hard coding the option keys.

Source code in winidjango/src/commands/base/base.py 
38
39
40
41
42
43
44
45
46
47
48
class Options:
    """Just a container class for hard coding the option keys."""

    DRY_RUN = "dry_run"
    FORCE = "force"
    DELETE = "delete"
    YES = "yes"
    TIMEOUT = "timeout"
    BATCH_SIZE = "batch_size"
    THREADS = "threads"
    PROCESSES = "processes"
add_arguments(parser)

Configure command-line arguments for the command.

Adds common base arguments (dry-run, force, delete, timeout, batching and concurrency options) and then delegates to :meth:add_command_arguments for command-specific options.

Parameters:

Name Type Description Default
parser ArgumentParser

The argument parser passed by Django.

 required
Source code in winidjango/src/commands/base/base.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def add_arguments(self, parser: ArgumentParser) -> None:
    """Configure command-line arguments for the command.

    Adds common base arguments (dry-run, force, delete, timeout,
    batching and concurrency options) and then delegates to
    :meth:`add_command_arguments` for command-specific options.

    Args:
        parser (ArgumentParser): The argument parser passed by Django.
    """
    # add base args that are used in most commands
    self.base_add_arguments(parser)

    # add additional args that are specific to the command
    self.add_command_arguments(parser)
add_command_arguments(parser) abstractmethod

Define command-specific arguments.

Implement this hook to add options and positional arguments that are specific to the concrete management command.

Parameters:

Name Type Description Default
parser ArgumentParser

The argument parser passed by Django.

 required
Source code in winidjango/src/commands/base/base.py 
125
126
127
128
129
130
131
132
133
134
@abstractmethod
def add_command_arguments(self, parser: ArgumentParser) -> None:
    """Define command-specific arguments.

    Implement this hook to add options and positional arguments that are
    specific to the concrete management command.

    Args:
        parser (ArgumentParser): The argument parser passed by Django.
    """
base_add_arguments(parser)

Add the project's standard command-line arguments to parser.

Parameters:

Name Type Description Default
parser ArgumentParser

The argument parser passed by Django.

 required
Source code in winidjango/src/commands/base/base.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def base_add_arguments(self, parser: ArgumentParser) -> None:
    """Add the project's standard command-line arguments to ``parser``.

    Args:
        parser (ArgumentParser): The argument parser passed by Django.
    """
    parser.add_argument(
        f"--{self.Options.DRY_RUN}",
        action="store_true",
        help="Show what would be done without actually executing the changes",
    )

    parser.add_argument(
        f"--{self.Options.FORCE}",
        action="store_true",
        help="Force an action in a command",
    )

    parser.add_argument(
        f"--{self.Options.DELETE}",
        action="store_true",
        help="Deleting smth in a command",
    )

    parser.add_argument(
        f"--{self.Options.YES}",
        action="store_true",
        help="Answer yes to all prompts",
        default=False,
    )

    parser.add_argument(
        f"--{self.Options.TIMEOUT}",
        type=int,
        help="Timeout for a command",
        default=None,
    )

    parser.add_argument(
        f"--{self.Options.BATCH_SIZE}",
        type=int,
        default=None,
        help="Number of items to process in each batch",
    )

    parser.add_argument(
        f"--{self.Options.THREADS}",
        type=int,
        default=None,
        help="Number of threads to use for processing",
    )

    parser.add_argument(
        f"--{self.Options.PROCESSES}",
        type=int,
        default=None,
        help="Number of processes to use for processing",
    )
base_handle(*args, **options)

Perform common pre-processing for commands.

Stores the incoming arguments and options on the instance for use by :meth:handle_command and subclasses.

Parameters:

Name Type Description Default
*args Any

Positional arguments forwarded from Django.

 ()
**options Any

Parsed command-line options.

 {}
Source code in winidjango/src/commands/base/base.py 
150
151
152
153
154
155
156
157
158
159
160
161
def base_handle(self, *args: Any, **options: Any) -> None:
    """Perform common pre-processing for commands.

    Stores the incoming arguments and options on the instance for use
    by :meth:`handle_command` and subclasses.

    Args:
        *args: Positional arguments forwarded from Django.
        **options: Parsed command-line options.
    """
    self.args = args
    self.options = options
get_option(option)

Retrieve a parsed command option by key.

Parameters:

Name Type Description Default
option str

The option key to retrieve from self.options.

 required

Returns:

Name Type Description
Any Any

The value for the requested option. If the option is not present a KeyError will be raised (matching how Django exposes options in management commands).
Source code in winidjango/src/commands/base/base.py 
173
174
175
176
177
178
179
180
181
182
183
184
def get_option(self, option: str) -> Any:
    """Retrieve a parsed command option by key.

    Args:
        option (str): The option key to retrieve from ``self.options``.

    Returns:
        Any: The value for the requested option. If the option is not
            present a ``KeyError`` will be raised (matching how Django
            exposes options in management commands).
    """
    return self.options[option]
handle(*args, **options)

Orchestrate command execution.

Performs shared pre-processing by calling :meth:base_handle and then delegates to :meth:handle_command which must be implemented by subclasses.

Parameters:

Name Type Description Default
*args Any

Positional arguments forwarded from Django.

 ()
**options Any

Parsed command-line options.

 {}
Source code in winidjango/src/commands/base/base.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
def handle(self, *args: Any, **options: Any) -> None:
    """Orchestrate command execution.

    Performs shared pre-processing by calling :meth:`base_handle` and
    then delegates to :meth:`handle_command` which must be implemented
    by subclasses.

    Args:
        *args: Positional arguments forwarded from Django.
        **options: Parsed command-line options.
    """
    self.base_handle(*args, **options)
    self.handle_command()
handle_command() abstractmethod

Run the command-specific behavior.

This abstract hook should be implemented by concrete commands to perform the command's main work. Implementations should read self.args and self.options which were set in :meth:base_handle.

Source code in winidjango/src/commands/base/base.py 
163
164
165
166
167
168
169
170
171
@abstractmethod
def handle_command(self) -> None:
    """Run the command-specific behavior.

    This abstract hook should be implemented by concrete commands to
    perform the command's main work. Implementations should read
    ``self.args`` and ``self.options`` which were set in
    :meth:`base_handle`.
    """
import_data

Import command base class and utilities.

This module defines a reusable base command for importing tabular data into Django models. Implementations should provide a concrete source ingestion (for example, reading from CSV or an external API), a cleaning/normalization step implemented by a CleaningDF subclass, and mapping logic that groups cleaned data into model instances that can be bulk-created.

The base command centralizes the typical flow: 1. Read raw data (handle_import) 2. Wrap and clean the data using a CleaningDF subclass 3. Convert the cleaned frame into per-model bulks 4. Persist bulks using the project's bulk create helpers

Using this base class ensures a consistent import lifecycle and reduces duplicated boilerplate across different import implementations.

ImportDataBaseCommand

Bases: ABCBaseCommand

Abstract base for data-import Django management commands.

Subclasses must implement the ingestion, cleaning-class selection, and mapping of cleaned rows to Django model instances. The base implementation wires these pieces together and calls the project's bulk creation helper to persist the data.

Implementors typically only need to override the three abstract methods documented below.

Source code in winidjango/src/commands/import_data.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
class ImportDataBaseCommand(ABCBaseCommand):
    """Abstract base for data-import Django management commands.

    Subclasses must implement the ingestion, cleaning-class selection,
    and mapping of cleaned rows to Django model instances. The base
    implementation wires these pieces together and calls the project's
    bulk creation helper to persist the data.

    Implementors typically only need to override the three abstract
    methods documented below.
    """

    @abstractmethod
    def handle_import(self) -> pl.DataFrame:
        """Read raw data from the import source.

        This method should read data from whatever source the concrete
        command targets (files, remote APIs, etc.) and return it as a
        ``polars.DataFrame``. No cleaning should be performed here;
        cleaning is handled by the cleaning `CleaningDF` returned from
        ``get_cleaning_df_cls``.

        Returns:
            pl.DataFrame: Raw (uncleaned) tabular data to be cleaned and
                mapped to model instances.
        """

    @abstractmethod
    def get_cleaning_df_cls(self) -> type[CleaningDF]:
        """Return the `CleaningDF` subclass used to normalize the data.

        The returned class will be instantiated with the raw DataFrame
        returned from :meth:`handle_import` and must provide the
        transformations required to prepare data for mapping into model
        instances.

        Returns:
            type[CleaningDF]: A subclass of ``CleaningDF`` that performs
                the necessary normalization and validation.
        """

    @abstractmethod
    def get_bulks_by_model(
        self, df: pl.DataFrame
    ) -> dict[type[Model], Iterable[Model]]:
        """Map the cleaned DataFrame to model-instance bulks.

        The implementation should inspect the cleaned DataFrame and
        return a mapping where keys are Django model classes and values
        are iterables of unsaved model instances (or dataclass-like
        objects accepted by the project's bulk-creation utility).

        Args:
            df (pl.DataFrame): The cleaned and normalized DataFrame.

        Returns:
            dict[type[Model], Iterable[Model]]: Mapping from model classes
                to iterables of instances that should be created.
        """

    def handle_command(self) -> None:
        """Execute the full import lifecycle.

        This template method reads raw data via :meth:`handle_import`,
        wraps it with the cleaning class returned by
        :meth:`get_cleaning_df_cls` and then persists the resulting
        model bulks returned by :meth:`get_bulks_by_model`.
        """
        data_df = self.handle_import()

        cleaning_df_cls = self.get_cleaning_df_cls()
        self.cleaning_df = cleaning_df_cls(data_df)

        self.import_to_db()

    def import_to_db(self) -> None:
        """Persist prepared model bulks to the database.

        Calls the project's `bulk_create_bulks_in_steps` helper with the
        mapping returned from :meth:`get_bulks_by_model`.
        """
        bulks_by_model = self.get_bulks_by_model(df=self.cleaning_df.df)

        bulk_create_bulks_in_steps(bulks_by_model)
Options

Just a container class for hard coding the option keys.

Source code in winidjango/src/commands/base/base.py 
38
39
40
41
42
43
44
45
46
47
48
class Options:
    """Just a container class for hard coding the option keys."""

    DRY_RUN = "dry_run"
    FORCE = "force"
    DELETE = "delete"
    YES = "yes"
    TIMEOUT = "timeout"
    BATCH_SIZE = "batch_size"
    THREADS = "threads"
    PROCESSES = "processes"
add_arguments(parser)

Configure command-line arguments for the command.

Adds common base arguments (dry-run, force, delete, timeout, batching and concurrency options) and then delegates to :meth:add_command_arguments for command-specific options.

Parameters:

Name Type Description Default
parser ArgumentParser

The argument parser passed by Django.

 required
Source code in winidjango/src/commands/base/base.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def add_arguments(self, parser: ArgumentParser) -> None:
    """Configure command-line arguments for the command.

    Adds common base arguments (dry-run, force, delete, timeout,
    batching and concurrency options) and then delegates to
    :meth:`add_command_arguments` for command-specific options.

    Args:
        parser (ArgumentParser): The argument parser passed by Django.
    """
    # add base args that are used in most commands
    self.base_add_arguments(parser)

    # add additional args that are specific to the command
    self.add_command_arguments(parser)
add_command_arguments(parser) abstractmethod

Define command-specific arguments.

Implement this hook to add options and positional arguments that are specific to the concrete management command.

Parameters:

Name Type Description Default
parser ArgumentParser

The argument parser passed by Django.

 required
Source code in winidjango/src/commands/base/base.py 
125
126
127
128
129
130
131
132
133
134
@abstractmethod
def add_command_arguments(self, parser: ArgumentParser) -> None:
    """Define command-specific arguments.

    Implement this hook to add options and positional arguments that are
    specific to the concrete management command.

    Args:
        parser (ArgumentParser): The argument parser passed by Django.
    """
base_add_arguments(parser)

Add the project's standard command-line arguments to parser.

Parameters:

Name Type Description Default
parser ArgumentParser

The argument parser passed by Django.

 required
Source code in winidjango/src/commands/base/base.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def base_add_arguments(self, parser: ArgumentParser) -> None:
    """Add the project's standard command-line arguments to ``parser``.

    Args:
        parser (ArgumentParser): The argument parser passed by Django.
    """
    parser.add_argument(
        f"--{self.Options.DRY_RUN}",
        action="store_true",
        help="Show what would be done without actually executing the changes",
    )

    parser.add_argument(
        f"--{self.Options.FORCE}",
        action="store_true",
        help="Force an action in a command",
    )

    parser.add_argument(
        f"--{self.Options.DELETE}",
        action="store_true",
        help="Deleting smth in a command",
    )

    parser.add_argument(
        f"--{self.Options.YES}",
        action="store_true",
        help="Answer yes to all prompts",
        default=False,
    )

    parser.add_argument(
        f"--{self.Options.TIMEOUT}",
        type=int,
        help="Timeout for a command",
        default=None,
    )

    parser.add_argument(
        f"--{self.Options.BATCH_SIZE}",
        type=int,
        default=None,
        help="Number of items to process in each batch",
    )

    parser.add_argument(
        f"--{self.Options.THREADS}",
        type=int,
        default=None,
        help="Number of threads to use for processing",
    )

    parser.add_argument(
        f"--{self.Options.PROCESSES}",
        type=int,
        default=None,
        help="Number of processes to use for processing",
    )
base_handle(*args, **options)

Perform common pre-processing for commands.

Stores the incoming arguments and options on the instance for use by :meth:handle_command and subclasses.

Parameters:

Name Type Description Default
*args Any

Positional arguments forwarded from Django.

 ()
**options Any

Parsed command-line options.

 {}
Source code in winidjango/src/commands/base/base.py 
150
151
152
153
154
155
156
157
158
159
160
161
def base_handle(self, *args: Any, **options: Any) -> None:
    """Perform common pre-processing for commands.

    Stores the incoming arguments and options on the instance for use
    by :meth:`handle_command` and subclasses.

    Args:
        *args: Positional arguments forwarded from Django.
        **options: Parsed command-line options.
    """
    self.args = args
    self.options = options
get_bulks_by_model(df) abstractmethod

Map the cleaned DataFrame to model-instance bulks.

The implementation should inspect the cleaned DataFrame and return a mapping where keys are Django model classes and values are iterables of unsaved model instances (or dataclass-like objects accepted by the project's bulk-creation utility).

Parameters:

Name Type Description Default
df DataFrame

The cleaned and normalized DataFrame.

 required

Returns:

Type Description
dict[type[Model], Iterable[Model]]

dict[type[Model], Iterable[Model]]: Mapping from model classes to iterables of instances that should be created.
Source code in winidjango/src/commands/import_data.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
@abstractmethod
def get_bulks_by_model(
    self, df: pl.DataFrame
) -> dict[type[Model], Iterable[Model]]:
    """Map the cleaned DataFrame to model-instance bulks.

    The implementation should inspect the cleaned DataFrame and
    return a mapping where keys are Django model classes and values
    are iterables of unsaved model instances (or dataclass-like
    objects accepted by the project's bulk-creation utility).

    Args:
        df (pl.DataFrame): The cleaned and normalized DataFrame.

    Returns:
        dict[type[Model], Iterable[Model]]: Mapping from model classes
            to iterables of instances that should be created.
    """
get_cleaning_df_cls() abstractmethod

Return the CleaningDF subclass used to normalize the data.

The returned class will be instantiated with the raw DataFrame returned from :meth:handle_import and must provide the transformations required to prepare data for mapping into model instances.

Returns:

Type Description
type[CleaningDF]

type[CleaningDF]: A subclass of CleaningDF that performs the necessary normalization and validation.
Source code in winidjango/src/commands/import_data.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
@abstractmethod
def get_cleaning_df_cls(self) -> type[CleaningDF]:
    """Return the `CleaningDF` subclass used to normalize the data.

    The returned class will be instantiated with the raw DataFrame
    returned from :meth:`handle_import` and must provide the
    transformations required to prepare data for mapping into model
    instances.

    Returns:
        type[CleaningDF]: A subclass of ``CleaningDF`` that performs
            the necessary normalization and validation.
    """
get_option(option)

Retrieve a parsed command option by key.

Parameters:

Name Type Description Default
option str

The option key to retrieve from self.options.

 required

Returns:

Name Type Description
Any Any

The value for the requested option. If the option is not present a KeyError will be raised (matching how Django exposes options in management commands).
Source code in winidjango/src/commands/base/base.py 
173
174
175
176
177
178
179
180
181
182
183
184
def get_option(self, option: str) -> Any:
    """Retrieve a parsed command option by key.

    Args:
        option (str): The option key to retrieve from ``self.options``.

    Returns:
        Any: The value for the requested option. If the option is not
            present a ``KeyError`` will be raised (matching how Django
            exposes options in management commands).
    """
    return self.options[option]
handle(*args, **options)

Orchestrate command execution.

Performs shared pre-processing by calling :meth:base_handle and then delegates to :meth:handle_command which must be implemented by subclasses.

Parameters:

Name Type Description Default
*args Any

Positional arguments forwarded from Django.

 ()
**options Any

Parsed command-line options.

 {}
Source code in winidjango/src/commands/base/base.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
def handle(self, *args: Any, **options: Any) -> None:
    """Orchestrate command execution.

    Performs shared pre-processing by calling :meth:`base_handle` and
    then delegates to :meth:`handle_command` which must be implemented
    by subclasses.

    Args:
        *args: Positional arguments forwarded from Django.
        **options: Parsed command-line options.
    """
    self.base_handle(*args, **options)
    self.handle_command()
handle_command()

Execute the full import lifecycle.

This template method reads raw data via :meth:handle_import, wraps it with the cleaning class returned by :meth:get_cleaning_df_cls and then persists the resulting model bulks returned by :meth:get_bulks_by_model.

Source code in winidjango/src/commands/import_data.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def handle_command(self) -> None:
    """Execute the full import lifecycle.

    This template method reads raw data via :meth:`handle_import`,
    wraps it with the cleaning class returned by
    :meth:`get_cleaning_df_cls` and then persists the resulting
    model bulks returned by :meth:`get_bulks_by_model`.
    """
    data_df = self.handle_import()

    cleaning_df_cls = self.get_cleaning_df_cls()
    self.cleaning_df = cleaning_df_cls(data_df)

    self.import_to_db()
handle_import() abstractmethod

Read raw data from the import source.

This method should read data from whatever source the concrete command targets (files, remote APIs, etc.) and return it as a polars.DataFrame. No cleaning should be performed here; cleaning is handled by the cleaning CleaningDF returned from get_cleaning_df_cls.

Returns:

Type Description
DataFrame

pl.DataFrame: Raw (uncleaned) tabular data to be cleaned and mapped to model instances.
Source code in winidjango/src/commands/import_data.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
@abstractmethod
def handle_import(self) -> pl.DataFrame:
    """Read raw data from the import source.

    This method should read data from whatever source the concrete
    command targets (files, remote APIs, etc.) and return it as a
    ``polars.DataFrame``. No cleaning should be performed here;
    cleaning is handled by the cleaning `CleaningDF` returned from
    ``get_cleaning_df_cls``.

    Returns:
        pl.DataFrame: Raw (uncleaned) tabular data to be cleaned and
            mapped to model instances.
    """
import_to_db()

Persist prepared model bulks to the database.

Calls the project's bulk_create_bulks_in_steps helper with the mapping returned from :meth:get_bulks_by_model.

Source code in winidjango/src/commands/import_data.py 
109
110
111
112
113
114
115
116
117
def import_to_db(self) -> None:
    """Persist prepared model bulks to the database.

    Calls the project's `bulk_create_bulks_in_steps` helper with the
    mapping returned from :meth:`get_bulks_by_model`.
    """
    bulks_by_model = self.get_bulks_by_model(df=self.cleaning_df.df)

    bulk_create_bulks_in_steps(bulks_by_model)

db

Database utilities and common model helpers used in the project.

This package contains helpers for working with Django models, such as topological sorting of model classes according to foreign-key dependencies, a lightweight model hashing helper, and a project-wide BaseModel that adds common timestamp fields.

bulk

Utilities for performing bulk operations on Django models.

This module centralizes helpers used by importers and maintenance commands to create, update and delete large collections of model instances efficiently. It provides batching, optional concurrent execution, dependency-aware ordering and simulation helpers for previewing cascade deletions.

bulk_create_bulks_in_steps(bulk_by_class, step=STANDARD_BULK_SIZE)

Create multiple model-type bulks in dependency order.

The function topologically sorts the provided model classes so that models referenced by foreign keys are created before models that reference them. Each class' instances are created in batches using :func:bulk_create_in_steps.

Parameters:

Name Type Description Default
bulk_by_class dict[type[Model], Iterable[Model]]

Mapping from model class to iterable of instances to create.

 required
step int

Chunk size for each model's batched creation.

 STANDARD_BULK_SIZE

Returns:

Type Description
dict[type[Model], list[Model]]

Mapping from model class to list of created instances.
Source code in winidjango/src/db/bulk.py 
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
def bulk_create_bulks_in_steps(
    bulk_by_class: dict[type[Model], Iterable[Model]],
    step: int = STANDARD_BULK_SIZE,
) -> dict[type[Model], list[Model]]:
    """Create multiple model-type bulks in dependency order.

    The function topologically sorts the provided model classes so that
    models referenced by foreign keys are created before models that
    reference them. Each class' instances are created in batches using
    :func:`bulk_create_in_steps`.

    Args:
        bulk_by_class: Mapping from model class to iterable of instances to create.
        step: Chunk size for each model's batched creation.

    Returns:
        Mapping from model class to list of created instances.
    """
    # order the bulks in order of creation depending how they depend on each other
    models_ = list(bulk_by_class.keys())
    ordered_models = topological_sort_models(models=models_)

    results: dict[type[Model], list[Model]] = {}
    for model_ in ordered_models:
        bulk = bulk_by_class[model_]
        result = bulk_create_in_steps(model=model_, bulk=bulk, step=step)
        results[model_] = result

    return results
bulk_create_in_steps(model, bulk, step=STANDARD_BULK_SIZE)

Create objects in batches and return created instances.

Breaks bulk into chunks of size step and calls the project's batched bulk-create helper for each chunk. Execution is performed using the concurrent utility where configured for throughput.

Parameters:

Name Type Description Default
model type[TModel]

Django model class to create instances for.

 required
bulk Iterable[TModel]

Iterable of unsaved model instances.

 required
step int

Number of instances to create per chunk.

 STANDARD_BULK_SIZE

Returns:

Type Description
list[TModel]

List of created model instances (flattened across chunks).
Source code in winidjango/src/db/bulk.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def bulk_create_in_steps[TModel: Model](
    model: type[TModel],
    bulk: Iterable[TModel],
    step: int = STANDARD_BULK_SIZE,
) -> list[TModel]:
    """Create objects in batches and return created instances.

    Breaks ``bulk`` into chunks of size ``step`` and calls the project's
    batched bulk-create helper for each chunk. Execution is performed
    using the concurrent utility where configured for throughput.

    Args:
        model: Django model class to create instances for.
        bulk: Iterable of unsaved model instances.
        step: Number of instances to create per chunk.

    Returns:
        List of created model instances (flattened across chunks).
    """
    return cast(
        "list[TModel]",
        bulk_method_in_steps(model=model, bulk=bulk, step=step, mode=MODE_CREATE),
    )
bulk_delete(model, objs, **_)

Delete the provided objects and return Django's delete summary.

Accepts either a QuerySet or an iterable of model instances. When an iterable of instances is provided it is converted to a QuerySet by filtering on primary keys before calling QuerySet.delete().

Parameters:

Name Type Description Default
model type[Model]

Django model class.

 required
objs Iterable[Model]

Iterable of model instances or a QuerySet.

 required

Returns:

Type Description
tuple[int, dict[str, int]]

Tuple(total_deleted, per_model_counts) as returned by delete().
Source code in winidjango/src/db/bulk.py 
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
def bulk_delete(
    model: type[Model], objs: Iterable[Model], **_: Any
) -> tuple[int, dict[str, int]]:
    """Delete the provided objects and return Django's delete summary.

    Accepts either a QuerySet or an iterable of model instances. When an
    iterable of instances is provided it is converted to a QuerySet by
    filtering on primary keys before calling ``QuerySet.delete()``.

    Args:
        model: Django model class.
        objs: Iterable of model instances or a QuerySet.

    Returns:
        Tuple(total_deleted, per_model_counts) as returned by ``delete()``.
    """
    return model.objects.filter(pk__in=(obj.pk for obj in objs)).delete()
bulk_delete_in_steps(model, bulk, step=STANDARD_BULK_SIZE)

Delete objects in batches and return deletion statistics.

Each chunk is deleted using Django's QuerySet delete which returns a (count, per-model-counts) tuple. Results are aggregated across chunks and returned as a consolidated tuple.

Parameters:

Name Type Description Default
model type[TModel]

Django model class.

 required
bulk Iterable[TModel]

Iterable of model instances to delete.

 required
step int

Chunk size for deletions.

 STANDARD_BULK_SIZE

Returns:

Type Description
int

A tuple containing the total number of deleted objects and a
dict[str, int]

mapping from model label to deleted count (including cascaded
tuple[int, dict[str, int]]

deletions).
Source code in winidjango/src/db/bulk.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def bulk_delete_in_steps[TModel: Model](
    model: type[TModel], bulk: Iterable[TModel], step: int = STANDARD_BULK_SIZE
) -> tuple[int, dict[str, int]]:
    """Delete objects in batches and return deletion statistics.

    Each chunk is deleted using Django's QuerySet ``delete`` which
    returns a (count, per-model-counts) tuple. Results are aggregated
    across chunks and returned as a consolidated tuple.

    Args:
        model: Django model class.
        bulk: Iterable of model instances to delete.
        step: Chunk size for deletions.

    Returns:
        A tuple containing the total number of deleted objects and a
        mapping from model label to deleted count (including cascaded
        deletions).
    """
    return cast(
        "tuple[int, dict[str, int]]",
        bulk_method_in_steps(
            model=model,
            bulk=bulk,
            step=step,
            mode=MODE_DELETE,
        ),
    )
bulk_method_in_steps(model, bulk, step, mode, **kwargs) 
bulk_method_in_steps(
    model: type[TModel],
    bulk: Iterable[TModel],
    step: int,
    mode: Literal["create"],
    **kwargs: Any,
) -> list[TModel]

bulk_method_in_steps(
    model: type[TModel],
    bulk: Iterable[TModel],
    step: int,
    mode: Literal["update"],
    **kwargs: Any,
) -> int

bulk_method_in_steps(
    model: type[TModel],
    bulk: Iterable[TModel],
    step: int,
    mode: Literal["delete"],
    **kwargs: Any,
) -> tuple[int, dict[str, int]]

Run a batched bulk operation (create/update/delete) on bulk.

This wrapper warns if called from within an existing transaction and delegates actual work to :func:bulk_method_in_steps_atomic which is executed inside an atomic transaction. The return type depends on mode (see :mod:winidjango.src.db.bulk constants).

Parameters:

Name Type Description Default
model type[TModel]

Django model class to operate on.

 required
bulk Iterable[TModel]

Iterable of model instances.

 required
step int

Chunk size for processing.

 required
mode MODE_TYPES

One of 'create', 'update' or 'delete'.

 required
**kwargs Any

Additional keyword arguments forwarded to the underlying bulk methods (for example update_fields for updates).

 {}

Returns:

Type Description
int | tuple[int, dict[str, int]] | list[TModel]

For create: list of created instances.
int | tuple[int, dict[str, int]] | list[TModel]

For update: integer number of updated rows.
int | tuple[int, dict[str, int]] | list[TModel]

For delete: tuple(total_deleted, per_model_counts).
Source code in winidjango/src/db/bulk.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
def bulk_method_in_steps[TModel: Model](
    model: type[TModel],
    bulk: Iterable[TModel],
    step: int,
    mode: MODE_TYPES,
    **kwargs: Any,
) -> int | tuple[int, dict[str, int]] | list[TModel]:
    """Run a batched bulk operation (create/update/delete) on ``bulk``.

    This wrapper warns if called from within an existing transaction and
    delegates actual work to :func:`bulk_method_in_steps_atomic` which is
    executed inside an atomic transaction. The return type depends on
    ``mode`` (see :mod:`winidjango.src.db.bulk` constants).

    Args:
        model: Django model class to operate on.
        bulk: Iterable of model instances.
        step: Chunk size for processing.
        mode: One of ``'create'``, ``'update'`` or ``'delete'``.
        **kwargs: Additional keyword arguments forwarded to the underlying
            bulk methods (for example ``update_fields`` for updates).

    Returns:
        For ``create``: list of created instances.
        For ``update``: integer number of updated rows.
        For ``delete``: tuple(total_deleted, per_model_counts).
    """
    # check if we are inside a transaction.atomic block
    _in_atomic_block = transaction.get_connection().in_atomic_block
    if _in_atomic_block:
        logger.info(
            "BE CAREFUL USING BULK OPERATIONS INSIDE A BROADER TRANSACTION BLOCK. "
            "BULKING WITH BULKS THAT DEPEND ON EACH OTHER CAN CAUSE "
            "INTEGRITY ERRORS OR POTENTIAL OTHER ISSUES."
        )
    return bulk_method_in_steps_atomic(
        model=model, bulk=bulk, step=step, mode=mode, **kwargs
    )
bulk_method_in_steps_atomic(model, bulk, step, mode, **kwargs) 
bulk_method_in_steps_atomic(
    model: type[TModel],
    bulk: Iterable[TModel],
    step: int,
    mode: Literal["create"],
    **kwargs: Any,
) -> list[TModel]

bulk_method_in_steps_atomic(
    model: type[TModel],
    bulk: Iterable[TModel],
    step: int,
    mode: Literal["update"],
    **kwargs: Any,
) -> int

bulk_method_in_steps_atomic(
    model: type[TModel],
    bulk: Iterable[TModel],
    step: int,
    mode: Literal["delete"],
    **kwargs: Any,
) -> tuple[int, dict[str, int]]

Atomic implementation of the batched bulk operation.

This function is decorated with transaction.atomic so each call to this function runs in a database transaction. Note that nesting additional, outer atomic blocks that also include dependent bulk operations can cause integrity issues for operations that depend on each other's side-effects; callers should be careful about atomic decorator placement in higher-level code.

Parameters:

Name Type Description Default
model type[TModel]

Django model class.

 required
bulk Iterable[TModel]

Iterable of model instances.

 required
step int

Chunk size for processing.

 required
mode MODE_TYPES

One of 'create', 'update' or 'delete'.

 required
**kwargs Any

Forwarded to the underlying bulk method.

 {}

Returns:

Name Type Description
See int | tuple[int, dict[str, int]] | list[TModel]

func:bulk_method_in_steps for return value semantics.
Source code in winidjango/src/db/bulk.py 
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
@transaction.atomic
def bulk_method_in_steps_atomic[TModel: Model](
    model: type[TModel],
    bulk: Iterable[TModel],
    step: int,
    mode: MODE_TYPES,
    **kwargs: Any,
) -> int | tuple[int, dict[str, int]] | list[TModel]:
    """Atomic implementation of the batched bulk operation.

    This function is decorated with ``transaction.atomic`` so each call
    to this function runs in a database transaction. Note that nesting
    additional, outer atomic blocks that also include dependent bulk
    operations can cause integrity issues for operations that depend on
    each other's side-effects; callers should be careful about atomic
    decorator placement in higher-level code.

    Args:
        model: Django model class.
        bulk: Iterable of model instances.
        step: Chunk size for processing.
        mode: One of ``'create'``, ``'update'`` or ``'delete'``.
        **kwargs: Forwarded to the underlying bulk method.

    Returns:
        See :func:`bulk_method_in_steps` for return value semantics.
    """
    bulk_method = get_bulk_method(model=model, mode=mode, **kwargs)

    chunks = get_step_chunks(bulk=bulk, step=step)

    # multithreading significantly increases speed
    result = multithread_loop(
        process_function=bulk_method,
        process_args=chunks,
    )

    return flatten_bulk_in_steps_result(result=result, mode=mode)
bulk_update_in_steps(model, bulk, update_fields, step=STANDARD_BULK_SIZE)

Update objects in batches and return total updated count.

Parameters:

Name Type Description Default
model type[TModel]

Django model class.

 required
bulk Iterable[TModel]

Iterable of model instances to update (must have PKs set).

 required
update_fields list[str]

Fields to update on each instance when calling bulk_update.

 required
step int

Chunk size for batched updates.

 STANDARD_BULK_SIZE

Returns:

Type Description
int

Total number of rows updated across all chunks.
Source code in winidjango/src/db/bulk.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def bulk_update_in_steps[TModel: Model](
    model: type[TModel],
    bulk: Iterable[TModel],
    update_fields: list[str],
    step: int = STANDARD_BULK_SIZE,
) -> int:
    """Update objects in batches and return total updated count.

    Args:
        model: Django model class.
        bulk: Iterable of model instances to update (must have PKs set).
        update_fields: Fields to update on each instance when calling
            ``bulk_update``.
        step: Chunk size for batched updates.

    Returns:
        Total number of rows updated across all chunks.
    """
    return cast(
        "int",
        bulk_method_in_steps(
            model=model, bulk=bulk, step=step, mode=MODE_UPDATE, fields=update_fields
        ),
    )
flatten_bulk_in_steps_result(result, mode) 
flatten_bulk_in_steps_result(
    result: list[list[TModel]], mode: Literal["create"]
) -> list[TModel]

flatten_bulk_in_steps_result(
    result: list[int], mode: Literal["update"]
) -> int

flatten_bulk_in_steps_result(
    result: list[tuple[int, dict[str, int]]],
    mode: Literal["delete"],
) -> tuple[int, dict[str, int]]

Aggregate per-chunk results returned by concurrent bulk execution.

Depending on mode the function reduces a list of per-chunk results into a single consolidated return value:

  • create: flattens a list of lists into a single list of objects
  • update: sums integer counts returned per chunk
  • delete: aggregates (count, per-model-dict) tuples into a single total and combined per-model counts

Parameters:

Name Type Description Default
result list[int] | list[tuple[int, dict[str, int]]] | list[list[TModel]]

List of per-chunk results returned by the chunk function.

 required
mode str

One of the supported modes.

 required

Returns:

Type Description
int | tuple[int, dict[str, int]] | list[TModel]

Aggregated result corresponding to mode.
Source code in winidjango/src/db/bulk.py 
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
def flatten_bulk_in_steps_result[TModel: Model](
    result: list[int] | list[tuple[int, dict[str, int]]] | list[list[TModel]], mode: str
) -> int | tuple[int, dict[str, int]] | list[TModel]:
    """Aggregate per-chunk results returned by concurrent bulk execution.

    Depending on ``mode`` the function reduces a list of per-chunk
    results into a single consolidated return value:

    - ``create``: flattens a list of lists into a single list of objects
    - ``update``: sums integer counts returned per chunk
    - ``delete``: aggregates (count, per-model-dict) tuples into a single
      total and combined per-model counts

    Args:
        result: List of per-chunk results returned by the chunk function.
        mode: One of the supported modes.

    Returns:
        Aggregated result corresponding to ``mode``.
    """
    if mode == MODE_UPDATE:
        # formated as [1000, 1000, ...]
        # since django 4.2 bulk_update returns the count of updated objects
        result = cast("list[int]", result)
        return int(sum(result))
    if mode == MODE_DELETE:
        # formated as [(count, {model_name: count, model_cascade_name: count}), ...]
        # join the results to get the total count of deleted objects
        result = cast("list[tuple[int, dict[str, int]]]", result)
        total_count = 0
        count_sum_by_model: defaultdict[str, int] = defaultdict(int)
        for count_sum, count_by_model in result:
            total_count += count_sum
            for model_name, count in count_by_model.items():
                count_sum_by_model[model_name] += count
        return (total_count, dict(count_sum_by_model))
    if mode == MODE_CREATE:
        # formated as [[obj1, obj2, ...], [obj1, obj2, ...], ...]
        result = cast("list[list[TModel]]", result)
        return [item for sublist in result for item in sublist]

    msg = f"Invalid method. Must be one of {MODES}"
    raise ValueError(msg)
get_bulk_method(model, mode, **kwargs) 
get_bulk_method(
    model: type[Model],
    mode: Literal["create"],
    **kwargs: Any,
) -> Callable[[list[Model]], list[Model]]

get_bulk_method(
    model: type[Model],
    mode: Literal["update"],
    **kwargs: Any,
) -> Callable[[list[Model]], int]

get_bulk_method(
    model: type[Model],
    mode: Literal["delete"],
    **kwargs: Any,
) -> Callable[[list[Model]], tuple[int, dict[str, int]]]

Return a callable that performs the requested bulk operation on a chunk.

The returned function accepts a single argument (a list of model instances) and returns the per-chunk result for the chosen mode.

Parameters:

Name Type Description Default
model type[Model]

Django model class.

 required
mode MODE_TYPES

One of 'create', 'update' or 'delete'.

 required
**kwargs Any

Forwarded to the underlying ORM bulk methods.

 {}

Raises:

Type Description
ValueError

If mode is invalid.

Returns:

Type Description
Callable[[list[Model]], list[Model] | int | tuple[int, dict[str, int]]]

Callable that accepts a list of model instances and returns the
Callable[[list[Model]], list[Model] | int | tuple[int, dict[str, int]]]

result for that chunk.
Source code in winidjango/src/db/bulk.py 
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
def get_bulk_method(
    model: type[Model], mode: MODE_TYPES, **kwargs: Any
) -> Callable[[list[Model]], list[Model] | int | tuple[int, dict[str, int]]]:
    """Return a callable that performs the requested bulk operation on a chunk.

    The returned function accepts a single argument (a list of model
    instances) and returns the per-chunk result for the chosen mode.

    Args:
        model: Django model class.
        mode: One of ``'create'``, ``'update'`` or ``'delete'``.
        **kwargs: Forwarded to the underlying ORM bulk methods.

    Raises:
        ValueError: If ``mode`` is invalid.

    Returns:
        Callable that accepts a list of model instances and returns the
        result for that chunk.
    """
    bulk_method: Callable[[list[Model]], list[Model] | int | tuple[int, dict[str, int]]]
    if mode == MODE_CREATE:

        def bulk_create_chunk(chunk: list[Model]) -> list[Model]:
            return model.objects.bulk_create(objs=chunk, **kwargs)

        bulk_method = bulk_create_chunk
    elif mode == MODE_UPDATE:

        def bulk_update_chunk(chunk: list[Model]) -> int:
            return model.objects.bulk_update(objs=chunk, **kwargs)

        bulk_method = bulk_update_chunk
    elif mode == MODE_DELETE:

        def bulk_delete_chunk(chunk: list[Model]) -> tuple[int, dict[str, int]]:
            return bulk_delete(model=model, objs=chunk, **kwargs)

        bulk_method = bulk_delete_chunk
    else:
        msg = f"Invalid method. Must be one of {MODES}"
        raise ValueError(msg)

    return bulk_method
get_differences_between_bulks(bulk1, bulk2, fields)

Return differences and intersections between two bulks of the same model.

Instances are compared using :func:hash_model_instance over the provided fields. The function maintains the original ordering for returned lists so that callers can preserve deterministic ordering when applying diffs.

Parameters:

Name Type Description Default
bulk1 list[TModel1]

First list of model instances.

 required
bulk2 list[TModel2]

Second list of model instances.

 required
fields list[Field[Any, Any] | ForeignObjectRel | GenericForeignKey]

Fields to include when hashing instances.

 required

Raises:

Type Description
ValueError

If bulks are empty or contain different model types.

Returns:

Type Description
tuple[list[TModel1], list[TModel2], list[TModel1], list[TModel2]]

Four lists in the order: (in_1_not_2, in_2_not_1, in_both_from_1, in_both_from_2).
Source code in winidjango/src/db/bulk.py 
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
def get_differences_between_bulks[TModel1: Model, TModel2: Model](
    bulk1: list[TModel1],
    bulk2: list[TModel2],
    fields: "list[Field[Any, Any] | ForeignObjectRel | GenericForeignKey]",
) -> tuple[list[TModel1], list[TModel2], list[TModel1], list[TModel2]]:
    """Return differences and intersections between two bulks of the same model.

    Instances are compared using :func:`hash_model_instance` over the
    provided ``fields``. The function maintains the original ordering
    for returned lists so that callers can preserve deterministic
    ordering when applying diffs.

    Args:
        bulk1: First list of model instances.
        bulk2: Second list of model instances.
        fields: Fields to include when hashing instances.

    Raises:
        ValueError: If bulks are empty or contain different model types.

    Returns:
        Four lists in the order:
            (in_1_not_2, in_2_not_1, in_both_from_1, in_both_from_2).
    """
    if not bulk1 or not bulk2:
        return bulk1, bulk2, [], []

    if type(bulk1[0]) is not type(bulk2[0]):
        msg = "Both bulks must be of the same model type."
        raise ValueError(msg)

    hash_model_instance_with_fields = partial(
        hash_model_instance,
        fields=fields,
    )
    # Precompute hashes and map them directly to models in a single pass for both bulks
    hashes1 = list(map(hash_model_instance_with_fields, bulk1))
    hashes2 = list(map(hash_model_instance_with_fields, bulk2))

    # Convert keys to sets for difference operations
    set1, set2 = set(hashes1), set(hashes2)

    # Calculate differences between sets
    # Find differences and intersection with original order preserved
    # Important, we need to return the original objects that are the same in memory,
    # so in_1_not_2 and in_2_not_1
    in_1_not_2 = set1 - set2
    in_1_not_2_list = [
        model
        for model, hash_ in zip(bulk1, hashes1, strict=False)
        if hash_ in in_1_not_2
    ]

    in_2_not_1 = set2 - set1
    in_2_not_1_list = [
        model
        for model, hash_ in zip(bulk2, hashes2, strict=False)
        if hash_ in in_2_not_1
    ]

    in_1_and_2 = set1 & set2
    in_1_and_2_from_1 = [
        model
        for model, hash_ in zip(bulk1, hashes1, strict=False)
        if hash_ in in_1_and_2
    ]
    in_1_and_2_from_2 = [
        model
        for model, hash_ in zip(bulk2, hashes2, strict=False)
        if hash_ in in_1_and_2
    ]

    return in_1_not_2_list, in_2_not_1_list, in_1_and_2_from_1, in_1_and_2_from_2
get_step_chunks(bulk, step)

Yield consecutive chunks of at most step items from bulk.

The function yields a single-tuple containing the chunk (a list of model instances) because the concurrent execution helper expects a tuple of positional arguments for the target function.

Parameters:

Name Type Description Default
bulk Iterable[Model]

Iterable of model instances.

 required
step int

Maximum number of instances per yielded chunk.

 required

Yields:

Type Description
tuple[list[Model]]

Tuples where the first element is a list of model instances.
Source code in winidjango/src/db/bulk.py 
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
def get_step_chunks(
    bulk: Iterable[Model], step: int
) -> Generator[tuple[list[Model]], None, None]:
    """Yield consecutive chunks of at most ``step`` items from ``bulk``.

    The function yields a single-tuple containing the chunk (a list of
    model instances) because the concurrent execution helper expects a
    tuple of positional arguments for the target function.

    Args:
        bulk: Iterable of model instances.
        step: Maximum number of instances per yielded chunk.

    Yields:
        Tuples where the first element is a list of model instances.
    """
    bulk = iter(bulk)
    while True:
        chunk = list(islice(bulk, step))
        if not chunk:
            break
        yield (chunk,)  # bc concurrent_loop expects a tuple of args
multi_simulate_bulk_deletion(entries)

Simulate deletions for multiple model classes and merge results.

Runs :func:simulate_bulk_deletion for each provided model and returns a unified mapping of all models that would be deleted.

Parameters:

Name Type Description Default
entries dict[type[Model], list[Model]]

Mapping from model class to list of instances to simulate.

 required

Returns:

Type Description
dict[type[Model], set[Model]]

Mapping from model class to set of instances that would be deleted.
Source code in winidjango/src/db/bulk.py 
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
def multi_simulate_bulk_deletion(
    entries: dict[type[Model], list[Model]],
) -> dict[type[Model], set[Model]]:
    """Simulate deletions for multiple model classes and merge results.

    Runs :func:`simulate_bulk_deletion` for each provided model and
    returns a unified mapping of all models that would be deleted.

    Args:
        entries: Mapping from model class to list of instances to simulate.

    Returns:
        Mapping from model class to set of instances that would be deleted.
    """
    deletion_summaries = [
        simulate_bulk_deletion(model, entry) for model, entry in entries.items()
    ]
    # join the dicts to get the total count of deleted objects
    joined_deletion_summary: defaultdict[type[Model], set[Model]] = defaultdict(set)
    for deletion_summary in deletion_summaries:
        for model, objects in deletion_summary.items():
            joined_deletion_summary[model].update(objects)

    return dict(joined_deletion_summary)
simulate_bulk_deletion(model_class, entries)

Simulate Django's delete cascade and return affected objects.

Uses :class:django.db.models.deletion.Collector to determine which objects (including cascaded related objects) would be removed if the provided entries were deleted. No database writes are performed.

Parameters:

Name Type Description Default
model_class type[TModel]

Model class of the provided entries.

 required
entries list[TModel]

Instances to simulate deletion for.

 required

Returns:

Type Description
dict[type[Model], set[Model]]

Mapping from model class to set of instances that would be deleted.
Source code in winidjango/src/db/bulk.py 
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
def simulate_bulk_deletion[TModel: Model](
    model_class: type[TModel], entries: list[TModel]
) -> dict[type[Model], set[Model]]:
    """Simulate Django's delete cascade and return affected objects.

    Uses :class:`django.db.models.deletion.Collector` to determine which
    objects (including cascaded related objects) would be removed if the
    provided entries were deleted. No database writes are performed.

    Args:
        model_class: Model class of the provided entries.
        entries: Instances to simulate deletion for.

    Returns:
        Mapping from model class to set of instances that would be deleted.
    """
    if not entries:
        return {}

    # Initialize the Collector
    using = router.db_for_write(model_class)
    collector = Collector(using)

    # Collect deletion cascade for all entries
    collector.collect(entries)  # ty:ignore[invalid-argument-type]

    # Prepare the result dictionary
    deletion_summary: defaultdict[type[Model], set[Model]] = defaultdict(set)

    # Add normal deletes
    for model, objects in collector.data.items():
        deletion_summary[model].update(objects)  # objects is already iterable

    # Add fast deletes (explicitly expand querysets)
    for queryset in collector.fast_deletes:
        deletion_summary[queryset.model].update(list(queryset))

    return deletion_summary
fields

Utilities for inspecting Django model fields.

This module provides small helpers that make it easier to introspect Django model fields and metadata in a type-friendly way. The helpers are used across the project's database utilities to implement operations like topological sorting and deterministic hashing of model instances.

get_field_names(fields)

Return the name attribute for a list of Django field objects.

Parameters:

Name Type Description Default
fields list[Field | ForeignObjectRel | GenericForeignKey]

Field objects obtained from a model's _meta.get_fields().

 required

Returns:

Type Description
list[str]

list[str]: List of field names in the same order as fields.
Source code in winidjango/src/db/fields.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
def get_field_names(
    fields: "list[Field[Any, Any] | ForeignObjectRel | GenericForeignKey]",
) -> list[str]:
    """Return the ``name`` attribute for a list of Django field objects.

    Args:
        fields (list[Field | ForeignObjectRel | GenericForeignKey]):
            Field objects obtained from a model's ``_meta.get_fields()``.

    Returns:
        list[str]: List of field names in the same order as ``fields``.
    """
    return [field.name for field in fields]
get_fields(model)

Return all field objects for a Django model.

This wraps model._meta.get_fields() and is typed to include relationship fields so callers can handle both regular and related fields uniformly.

Parameters:

Name Type Description Default
model type[Model]

Django model class.

 required

Returns:

Type Description
list[Field[Any, Any] | ForeignObjectRel | GenericForeignKey]

list[Field | ForeignObjectRel | GenericForeignKey]: All field objects associated with the model.
Source code in winidjango/src/db/fields.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def get_fields[TModel: Model](
    model: type[TModel],
) -> "list[Field[Any, Any] | ForeignObjectRel | GenericForeignKey]":
    """Return all field objects for a Django model.

    This wraps ``model._meta.get_fields()`` and is typed to include
    relationship fields so callers can handle both regular and related
    fields uniformly.

    Args:
        model (type[Model]): Django model class.

    Returns:
        list[Field | ForeignObjectRel | GenericForeignKey]: All field
            objects associated with the model.
    """
    return get_model_meta(model).get_fields()
get_model_meta(model)

Return a model class' _meta options object.

This small wrapper exists to make typing clearer at call sites where the code needs the model Options object.

Parameters:

Name Type Description Default
model type[Model]

Django model class.

 required

Returns:

Name Type Description
Options Options[Model]

The model's _meta options object.
Source code in winidjango/src/db/fields.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
def get_model_meta(model: type[Model]) -> "Options[Model]":
    """Return a model class' ``_meta`` options object.

    This small wrapper exists to make typing clearer at call sites where
    the code needs the model Options object.

    Args:
        model (type[Model]): Django model class.

    Returns:
        Options: The model's ``_meta`` options object.
    """
    return model._meta  # noqa: SLF001
models

Database utilities and lightweight model helpers.

This module provides helpers used across the project when manipulating Django models: ordering models by foreign-key dependencies, creating a deterministic hash for unsaved instances, and a project-wide BaseModel that exposes common timestamp fields.

BaseModel

Bases: Model

Abstract base model containing common fields and helpers.

Concrete models can inherit from this class to get consistent created_at and updated_at timestamp fields and convenient string representations.

Source code in winidjango/src/db/models.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
class BaseModel(Model):
    """Abstract base model containing common fields and helpers.

    Concrete models can inherit from this class to get consistent
    ``created_at`` and ``updated_at`` timestamp fields and convenient
    string representations.
    """

    created_at: DateTimeField[datetime, datetime] = DateTimeField(auto_now_add=True)
    updated_at: DateTimeField[datetime, datetime] = DateTimeField(auto_now=True)

    class Meta:
        """Mark the model as abstract."""

        # abstract does not inherit in children
        abstract = True

    def __str__(self) -> str:
        """Return a concise human-readable representation.

        The default shows the model class name and primary key which is
        useful for logging and interactive debugging.

        Returns:
            str: Short representation, e.g. ``MyModel(123)``.
        """
        return f"{self.__class__.__name__}({self.pk})"

    def __repr__(self) -> str:
        """Base representation of a model."""
        return str(self)

    @property
    def meta(self) -> "Options[Self]":
        """Return the model's ``_meta`` options object.

        This property is a small convenience wrapper used to make access
        sites slightly more explicit in code and improve typing in callers.
        """
        return self._meta
meta property

Return the model's _meta options object.

This property is a small convenience wrapper used to make access sites slightly more explicit in code and improve typing in callers.

Meta

Mark the model as abstract.

Source code in winidjango/src/db/models.py 
130
131
132
133
134
class Meta:
    """Mark the model as abstract."""

    # abstract does not inherit in children
    abstract = True
__repr__()

Base representation of a model.

Source code in winidjango/src/db/models.py 
147
148
149
def __repr__(self) -> str:
    """Base representation of a model."""
    return str(self)
__str__()

Return a concise human-readable representation.

The default shows the model class name and primary key which is useful for logging and interactive debugging.

Returns:

Name Type Description
str str

Short representation, e.g. MyModel(123).
Source code in winidjango/src/db/models.py 
136
137
138
139
140
141
142
143
144
145
def __str__(self) -> str:
    """Return a concise human-readable representation.

    The default shows the model class name and primary key which is
    useful for logging and interactive debugging.

    Returns:
        str: Short representation, e.g. ``MyModel(123)``.
    """
    return f"{self.__class__.__name__}({self.pk})"
hash_model_instance(instance, fields)

Compute a deterministic hash for a model instance.

The function returns a hash suitable for comparing unsaved model instances by their field values. If the instance has a primary key (instance.pk) that key is hashed and returned immediately; this keeps comparisons cheap for persisted objects.

Parameters:

Name Type Description Default
instance Model

The Django model instance to hash.

 required
fields list[Field | ForeignObjectRel | GenericForeignKey]

Field objects that should be included when computing the hash.

 required

Returns:

Name Type Description
int int

Deterministic integer hash of the instance. For persisted instances this is hash(instance.pk).
Notes
  • The returned hash is intended for heuristic comparisons (e.g. deduplication in import pipelines) and is not cryptographically secure. Use with care when relying on absolute uniqueness.
Source code in winidjango/src/db/models.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def hash_model_instance(
    instance: Model,
    fields: "list[Field[Any, Any] | ForeignObjectRel | GenericForeignKey]",
) -> int:
    """Compute a deterministic hash for a model instance.

    The function returns a hash suitable for comparing unsaved model
    instances by their field values. If the instance has a primary key
    (``instance.pk``) that key is hashed and returned immediately; this
    keeps comparisons cheap for persisted objects.

    Args:
        instance (Model): The Django model instance to hash.
        fields (list[Field | ForeignObjectRel | GenericForeignKey]):
            Field objects that should be included when computing the hash.

    Returns:
        int: Deterministic integer hash of the instance. For persisted
            instances this is ``hash(instance.pk)``.

    Notes:
        - The returned hash is intended for heuristic comparisons (e.g.
          deduplication in import pipelines) and is not cryptographically
          secure. Use with care when relying on absolute uniqueness.
    """
    if instance.pk:
        return hash(instance.pk)

    field_names = get_field_names(fields)
    model_dict = model_to_dict(instance, fields=field_names)
    sorted_dict = dict(sorted(model_dict.items()))
    values = (type(instance), tuple(sorted_dict.items()))
    return hash(values)
topological_sort_models(models)

Sort Django models in dependency order using topological sorting.

Analyzes foreign key relationships between Django models and returns them in an order where dependencies come before dependents. This ensures that when performing operations like bulk creation or deletion, models are processed in the correct order to avoid foreign key constraint violations.

The function uses Python's graphlib.TopologicalSorter to perform the sorting based on ForeignKey relationships between the provided models. Only relationships between models in the input list are considered.

Parameters:

Name Type Description Default
models list[type[Model]]

A list of Django model classes to sort based on their foreign key dependencies.

 required

Returns:

Type Description
list[type[Model]]

list[type[Model]]: The input models sorted in dependency order, where models that are referenced by foreign keys appear before models that reference them. Self-referential relationships are ignored.

Raises:

Type Description
CycleError

If there are circular dependencies between models that cannot be resolved.
Example
Assuming Author model has no dependencies
and Book model has ForeignKey to Author

models = [Book, Author] sorted_models = topological_sort_models(models) sorted_models [, ]

Note
  • Only considers ForeignKey relationships, not other field types
  • Self-referential foreign keys are ignored to avoid self-loops
  • Only relationships between models in the input list are considered
Source code in winidjango/src/db/models.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def topological_sort_models(
    models: list[type[Model]],
) -> list[type[Model]]:
    """Sort Django models in dependency order using topological sorting.

    Analyzes foreign key relationships between Django models and returns them
    in an order where dependencies come before dependents. This ensures that
    when performing operations like bulk creation or deletion, models are
    processed in the correct order to avoid foreign key constraint violations.

    The function uses Python's graphlib.TopologicalSorter to perform the sorting
    based on ForeignKey relationships between the provided models. Only
    relationships between models in the input list are considered.

    Args:
        models (list[type[Model]]): A list of Django model classes to sort
            based on their foreign key dependencies.

    Returns:
        list[type[Model]]: The input models sorted in dependency order, where
            models that are referenced by foreign keys appear before models
            that reference them. Self-referential relationships are ignored.

    Raises:
        graphlib.CycleError: If there are circular dependencies between models
            that cannot be resolved.

    Example:
        >>> # Assuming Author model has no dependencies
        >>> # and Book model has ForeignKey to Author
        >>> models = [Book, Author]
        >>> sorted_models = topological_sort_models(models)
        >>> sorted_models
        [<class 'Author'>, <class 'Book'>]

    Note:
        - Only considers ForeignKey relationships, not other field types
        - Self-referential foreign keys are ignored to avoid self-loops
        - Only relationships between models in the input list are considered
    """
    ts: TopologicalSorter[type[Model]] = TopologicalSorter()

    for model in models:
        deps = {
            field.related_model
            for field in get_fields(model)
            if isinstance(field, ForeignKey)
            and isinstance(field.related_model, type)
            and field.related_model in models
            and field.related_model is not model
        }
        ts.add(model, *deps)

    return list(ts.static_order())
sql

Low-level helper to execute raw SQL against Django's database.

This module exposes :func:execute_sql which runs a parameterized SQL query using Django's database connection and returns column names and rows. It is intended for one-off queries where ORM abstractions are insufficient or when reading complex reports from the database.

The helper uses Django's connection cursor context manager to ensure resources are cleaned up correctly. Results are fetched into memory so avoid using it for very large result sets.

execute_sql(sql, params=None)

Execute a SQL statement and return column names and rows.

Parameters:

Name Type Description Default
sql str

SQL statement possibly containing named placeholders (%(name)s) for database binding.

 required
params dict[str, Any] | None

Optional mapping of parameters to bind to the query.

 None

Returns:

Type Description
list[str]

Tuple[List[str], List[Tuple[Any, ...]]]: A tuple where the first
list[tuple[Any, ...]]

element is the list of column names (empty list if the statement
tuple[list[str], list[tuple[Any, ...]]]

returned no rows) and the second element is a list of row tuples.

Raises:

Type Description
Error

Propagates underlying database errors raised by Django's database backend.
Source code in winidjango/src/db/sql.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def execute_sql(
    sql: str, params: dict[str, Any] | None = None
) -> tuple[list[str], list[tuple[Any, ...]]]:
    """Execute a SQL statement and return column names and rows.

    Args:
        sql (str): SQL statement possibly containing named placeholders
            (``%(name)s``) for database binding.
        params (dict[str, Any] | None): Optional mapping of parameters to
            bind to the query.

    Returns:
        Tuple[List[str], List[Tuple[Any, ...]]]: A tuple where the first
        element is the list of column names (empty list if the statement
        returned no rows) and the second element is a list of row tuples.

    Raises:
        django.db.Error: Propagates underlying database errors raised by
            Django's database backend.
    """
    with connection.cursor() as cursor:
        cursor.execute(sql=sql, params=params)
        rows = cursor.fetchall()
        column_names = (
            [col[0] for col in cursor.description] if cursor.description else []
        )

    return column_names, rows