Skip to content

System Internal Modules

This section documents the lower-level system modules used to compose site settings, editable translations, profile models, fixture import helpers, and Redis sync behavior.

Context processors

codex_django.system.context_processors

Context processors for project-level system data.

This module injects site settings and editable static content into templates, using the configured Django models and Redis-backed managers when available.

Classes

Functions

site_settings(request)

Expose cached site settings in templates via a safe proxy object.

The processor reads the configured singleton settings model, loads its cached representation through the Redis-backed manager, and injects the result into templates as site_settings. Missing configuration or cache failures degrade to an empty :class:SettingsProxy.

Parameters:

Name Type Description Default
request HttpRequest

Incoming Django request. The request itself is unused but kept to satisfy the Django context processor contract.

 required

Returns:

Name Type Description
dict[str, Any]

A template context mapping with the site_settings key containing
a dict[str, Any]

class:SettingsProxy instance.
Source code in src/codex_django/system/context_processors.py 
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
46
47
def site_settings(request: HttpRequest) -> dict[str, Any]:
    """Expose cached site settings in templates via a safe proxy object.

    The processor reads the configured singleton settings model, loads its
    cached representation through the Redis-backed manager, and injects the
    result into templates as ``site_settings``. Missing configuration or
    cache failures degrade to an empty :class:`SettingsProxy`.

    Args:
        request: Incoming Django request. The request itself is unused but
            kept to satisfy the Django context processor contract.

    Returns:
        A template context mapping with the ``site_settings`` key containing
        a :class:`SettingsProxy` instance.
    """
    model_path = getattr(settings, "CODEX_SITE_SETTINGS_MODEL", None)
    if not model_path:
        return {"site_settings": SettingsProxy({})}

    try:
        model = apps.get_model(model_path)
        manager = get_site_settings_manager()
        data = manager.load_cached(model)
        return {"site_settings": SettingsProxy(data)}
    except Exception as e:
        log.warning("site_settings context processor failed: %s", e)
        return {"site_settings": SettingsProxy({})}

static_content(request)

Expose static translation content as a simple template dictionary.

The processor reads the configured translation model and materializes all key -> content pairs into the template context as static_content. Missing configuration or query failures degrade to an empty dictionary.

Parameters:

Name Type Description Default
request HttpRequest

Incoming Django request. The request itself is unused but kept to satisfy the Django context processor contract.

 required

Returns:

Type Description
dict[str, Any]

A template context mapping with the static_content key.
Source code in src/codex_django/system/context_processors.py 
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
def static_content(request: HttpRequest) -> dict[str, Any]:
    """Expose static translation content as a simple template dictionary.

    The processor reads the configured translation model and materializes all
    ``key -> content`` pairs into the template context as ``static_content``.
    Missing configuration or query failures degrade to an empty dictionary.

    Args:
        request: Incoming Django request. The request itself is unused but
            kept to satisfy the Django context processor contract.

    Returns:
        A template context mapping with the ``static_content`` key.
    """
    model_path = getattr(settings, "CODEX_STATIC_TRANSLATION_MODEL", None)
    if not model_path:
        return {"static_content": {}}

    try:
        model = apps.get_model(model_path)
        manager = get_static_content_manager()
        data = manager.load_cached(model)
        return {"static_content": data}
    except Exception as e:
        log.warning("static_content context processor failed: %s", e)
        return {"static_content": {}}

Settings mixins

codex_django.system.mixins.settings

Reusable mixins for building a project-specific site settings model.

Combine :class:AbstractSiteSettings with the desired field mixins inside the target project's system app, then point settings.CODEX_SITE_SETTINGS_MODEL to the concrete model. Instances are serialized to Redis automatically through :class:SiteSettingsSyncMixin.

Classes

SiteSettingsSyncMixin

Bases: LifecycleModelMixin, Model

Provide Redis synchronization helpers for site settings models.

Notes

Projects usually do not instantiate this mixin directly. Instead it is inherited through :class:AbstractSiteSettings.

Source code in src/codex_django/system/mixins/settings.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
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
class SiteSettingsSyncMixin(LifecycleModelMixin, models.Model):
    """Provide Redis synchronization helpers for site settings models.

    Notes:
        Projects usually do not instantiate this mixin directly. Instead it
        is inherited through :class:`AbstractSiteSettings`.
    """

    class Meta:
        abstract = True

    # Redis manager class used for sync
    redis_manager_class = DjangoSiteSettingsManager

    def to_dict(self) -> dict[str, Any]:
        """Serialize concrete model fields into a Redis-friendly mapping.

        Returns:
            A dictionary containing concrete non-relational field values.
            File fields are converted to their public URL when available.
        """
        data: dict[str, Any] = {}
        for field in self._meta.get_fields():
            if field.concrete and not field.many_to_many and not field.one_to_many:
                if field.name in ["id", "pk"]:
                    continue

                value = getattr(self, field.name)
                # Handle files (store the URL)
                if isinstance(field, models.FileField) and value:
                    try:
                        data[field.name] = value.url
                    except ValueError:
                        data[field.name] = None
                else:
                    data[field.name] = value
        return data

    @hook(AFTER_SAVE)  # type: ignore[untyped-decorator]
    def sync_to_redis(self) -> None:
        """Hook to automatically update Redis cache upon saving.

        In DEBUG mode Redis sync is skipped unless CODEX_REDIS_ENABLED=True.
        This allows local development without a running Redis instance.
        """
        from django.conf import settings

        if settings.DEBUG and not getattr(settings, "CODEX_REDIS_ENABLED", False):
            return

        from codex_django.core.redis.managers.settings import get_site_settings_manager

        manager = get_site_settings_manager()
        manager.save_instance(self)
Functions
to_dict()

Serialize concrete model fields into a Redis-friendly mapping.

Returns:

Type Description
dict[str, Any]

A dictionary containing concrete non-relational field values.
dict[str, Any]

File fields are converted to their public URL when available.
Source code in src/codex_django/system/mixins/settings.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def to_dict(self) -> dict[str, Any]:
    """Serialize concrete model fields into a Redis-friendly mapping.

    Returns:
        A dictionary containing concrete non-relational field values.
        File fields are converted to their public URL when available.
    """
    data: dict[str, Any] = {}
    for field in self._meta.get_fields():
        if field.concrete and not field.many_to_many and not field.one_to_many:
            if field.name in ["id", "pk"]:
                continue

            value = getattr(self, field.name)
            # Handle files (store the URL)
            if isinstance(field, models.FileField) and value:
                try:
                    data[field.name] = value.url
                except ValueError:
                    data[field.name] = None
            else:
                data[field.name] = value
    return data
sync_to_redis()

Hook to automatically update Redis cache upon saving.

In DEBUG mode Redis sync is skipped unless CODEX_REDIS_ENABLED=True. This allows local development without a running Redis instance.

Source code in src/codex_django/system/mixins/settings.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@hook(AFTER_SAVE)  # type: ignore[untyped-decorator]
def sync_to_redis(self) -> None:
    """Hook to automatically update Redis cache upon saving.

    In DEBUG mode Redis sync is skipped unless CODEX_REDIS_ENABLED=True.
    This allows local development without a running Redis instance.
    """
    from django.conf import settings

    if settings.DEBUG and not getattr(settings, "CODEX_REDIS_ENABLED", False):
        return

    from codex_django.core.redis.managers.settings import get_site_settings_manager

    manager = get_site_settings_manager()
    manager.save_instance(self)

SiteContactSettingsMixin

Bases: Model

Add contact information fields to a site settings model.

Notes

Use this mixin when the project needs a single editable source for phone, email, address, and public contact metadata.

Admin

fieldsets: ( _("Contact Information"), { "fields": ( "phone", "email", "address_street", "address_locality", "address_postal_code", "contact_person", "working_hours", ), }, )

Source code in src/codex_django/system/mixins/settings.py 
 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
class SiteContactSettingsMixin(models.Model):
    """Add contact information fields to a site settings model.

    Notes:
        Use this mixin when the project needs a single editable source for
        phone, email, address, and public contact metadata.

    Admin:
        fieldsets: (
            _("Contact Information"),
            {
                "fields": (
                    "phone",
                    "email",
                    "address_street",
                    "address_locality",
                    "address_postal_code",
                    "contact_person",
                    "working_hours",
                ),
            },
        )
    """

    phone = models.CharField(_("Phone"), max_length=50, blank=True)
    email = models.EmailField(_("Email"), blank=True)
    address_street = models.CharField(_("Street Address"), max_length=255, blank=True)
    address_locality = models.CharField(_("City/Locality"), max_length=255, blank=True)
    address_postal_code = models.CharField(_("Postal Code"), max_length=10, blank=True)

    contact_person = models.CharField(_("Contact Person"), max_length=255, blank=True)
    working_hours = models.TextField(_("Working Hours"), blank=True)

    class Meta:
        abstract = True

SiteGeoSettingsMixin

Bases: Model

Add map and geographic metadata fields to a site settings model.

Notes

These fields are useful for map embeds, structured data, and contact pages that need coordinates or provider links.

Admin

fieldsets: (_("Geography & Map"), {"fields": ("google_maps_link", "latitude", "longitude"), "classes": ("collapse",)})

Source code in src/codex_django/system/mixins/settings.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
class SiteGeoSettingsMixin(models.Model):
    """Add map and geographic metadata fields to a site settings model.

    Notes:
        These fields are useful for map embeds, structured data, and contact
        pages that need coordinates or provider links.

    Admin:
        fieldsets:
            (_("Geography & Map"), {"fields": ("google_maps_link", "latitude", "longitude"), "classes": ("collapse",)})
    """

    google_maps_link = models.URLField(_("Google Maps Link"), blank=True, max_length=500)
    latitude = models.CharField(_("Latitude"), max_length=50, blank=True)
    longitude = models.CharField(_("Longitude"), max_length=50, blank=True)

    class Meta:
        abstract = True

SiteSocialSettingsMixin

Bases: Model

Add social network URL fields to a site settings model.

Notes

The mixin is intentionally broad so projects can expose only the subset of networks they actually render in templates.

Admin

fieldsets: ( _("Social Networks"), { "fields": ( "instagram_url", "facebook_url", "telegram_url", "whatsapp_url", "youtube_url", "linkedin_url", "tiktok_url", "twitter_url", "pinterest_url", ), "classes": ("collapse",), }, )

Source code in src/codex_django/system/mixins/settings.py 
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
class SiteSocialSettingsMixin(models.Model):
    """Add social network URL fields to a site settings model.

    Notes:
        The mixin is intentionally broad so projects can expose only the
        subset of networks they actually render in templates.

    Admin:
        fieldsets: (
            _("Social Networks"),
            {
                "fields": (
                    "instagram_url",
                    "facebook_url",
                    "telegram_url",
                    "whatsapp_url",
                    "youtube_url",
                    "linkedin_url",
                    "tiktok_url",
                    "twitter_url",
                    "pinterest_url",
                ),
                "classes": ("collapse",),
            },
        )
    """

    instagram_url = models.URLField(_("Instagram URL"), blank=True)
    facebook_url = models.URLField(_("Facebook URL"), blank=True)
    telegram_url = models.URLField(_("Telegram URL"), blank=True)
    whatsapp_url = models.URLField(_("WhatsApp URL"), blank=True)
    youtube_url = models.URLField(_("YouTube URL"), blank=True)
    linkedin_url = models.URLField(_("LinkedIn URL"), blank=True)
    tiktok_url = models.URLField(_("TikTok URL"), blank=True)
    twitter_url = models.URLField(_("Twitter (X) URL"), blank=True)
    pinterest_url = models.URLField(_("Pinterest URL"), blank=True)

    class Meta:
        abstract = True

SiteMarketingSettingsMixin

Bases: Model

Add analytics and marketing tracking identifiers.

Notes

These values are typically consumed by template includes that inject analytics and pixel scripts conditionally.

Admin

fieldsets: ( _("Marketing & Analytics"), { "fields": ( "google_analytics_id", "google_tag_manager_id", "facebook_pixel_id", "tiktok_pixel_id", ), "classes": ("collapse",), }, )

Source code in src/codex_django/system/mixins/settings.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
class SiteMarketingSettingsMixin(models.Model):
    """Add analytics and marketing tracking identifiers.

    Notes:
        These values are typically consumed by template includes that inject
        analytics and pixel scripts conditionally.

    Admin:
        fieldsets: (
            _("Marketing & Analytics"),
            {
                "fields": (
                    "google_analytics_id",
                    "google_tag_manager_id",
                    "facebook_pixel_id",
                    "tiktok_pixel_id",
                ),
                "classes": ("collapse",),
            },
        )
    """

    google_analytics_id = models.CharField(_("Google Analytics ID"), max_length=50, blank=True)
    google_tag_manager_id = models.CharField(_("Google Tag Manager ID"), max_length=50, blank=True)
    facebook_pixel_id = models.CharField(_("Facebook Pixel ID"), max_length=50, blank=True)
    tiktok_pixel_id = models.CharField(_("TikTok Pixel ID"), max_length=50, blank=True)

    class Meta:
        abstract = True

SiteTechnicalSettingsMixin

Bases: Model

Add operational flags and raw script injection fields.

Notes

This mixin is aimed at operational toggles and controlled script injection, not arbitrary long-form content storage.

Admin

fieldsets: ( _("Technical Settings"), { "fields": ("app_mode_enabled", "maintenance_mode", "head_scripts", "body_scripts"), "classes": ("collapse",), }, )

Source code in src/codex_django/system/mixins/settings.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
class SiteTechnicalSettingsMixin(models.Model):
    """Add operational flags and raw script injection fields.

    Notes:
        This mixin is aimed at operational toggles and controlled script
        injection, not arbitrary long-form content storage.

    Admin:
        fieldsets: (
            _("Technical Settings"),
            {
                "fields": ("app_mode_enabled", "maintenance_mode", "head_scripts", "body_scripts"),
                "classes": ("collapse",),
            },
        )
    """

    app_mode_enabled = models.BooleanField(_("App Mode Enabled"), default=False)
    maintenance_mode = models.BooleanField(_("Maintenance Mode"), default=False)
    head_scripts = models.TextField(_("Scripts in <head>"), blank=True)
    body_scripts = models.TextField(_("Scripts in <body>"), blank=True)

    class Meta:
        abstract = True

SiteEmailIdentityMixin

Bases: Model

Add outgoing-mail identity fields (From address, sender name, reply-to).

Transport (host/port/user/password/TLS) is configured via Django's EMAIL_* settings and the .env file, not through this model. These fields only describe who a notification is from, so that a site owner can safely adjust branding from the cabinet UI without touching SMTP credentials.

Admin

fieldsets: ( _("Email Identity"), { "fields": ("email_from", "email_sender_name", "email_reply_to"), "classes": ("collapse",), }, )

Source code in src/codex_django/system/mixins/settings.py 
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
class SiteEmailIdentityMixin(models.Model):
    """Add outgoing-mail *identity* fields (From address, sender name, reply-to).

    Transport (host/port/user/password/TLS) is configured via Django's
    ``EMAIL_*`` settings and the ``.env`` file, not through this model. These
    fields only describe **who** a notification is from, so that a site owner
    can safely adjust branding from the cabinet UI without touching SMTP
    credentials.

    Admin:
        fieldsets: (
            _("Email Identity"),
            {
                "fields": ("email_from", "email_sender_name", "email_reply_to"),
                "classes": ("collapse",),
            },
        )
    """

    email_from = models.EmailField(_("From Email"), blank=True)
    email_sender_name = models.CharField(_("Sender Name"), max_length=255, blank=True)
    email_reply_to = models.EmailField(_("Reply-To"), blank=True)

    class Meta:
        abstract = True

AbstractSiteSettings

Bases: SiteSettingsSyncMixin, Model

Base abstract site settings model with Redis synchronization only.

Notes

Subclass this model and compose it with the field mixins from this module to build the concrete settings model for a project.

Admin

Typically registered as a singleton-like model combined with the desired fieldset groups from the selected mixins.

Source code in src/codex_django/system/mixins/settings.py 
256
257
258
259
260
261
262
263
264
265
266
267
268
269
class AbstractSiteSettings(SiteSettingsSyncMixin, models.Model):
    """Base abstract site settings model with Redis synchronization only.

    Notes:
        Subclass this model and compose it with the field mixins from this
        module to build the concrete settings model for a project.

    Admin:
        Typically registered as a singleton-like model combined with the
        desired fieldset groups from the selected mixins.
    """

    class Meta:
        abstract = True

SEO mixins

codex_django.system.mixins.seo

Reusable SEO model mixins for named static pages.

Examples:

Declare a project SEO model in the local system app::

from codex_django.system.mixins.seo import AbstractStaticPageSeo

class StaticPageSeo(AbstractStaticPageSeo):
    pass

Classes

AbstractStaticPageSeo

Bases: TimestampMixin, SeoMixin

Store SEO metadata for named static pages with automatic cache invalidation.

Saving the model invalidates the Redis entry managed by :func:codex_django.core.redis.managers.seo.get_seo_redis_manager.

Admin

list_display: ("page_key", "seo_title", "updated_at") search_fields: ("page_key", "seo_title", "seo_description") readonly_fields: ("created_at", "updated_at")

Source code in src/codex_django/system/mixins/seo.py 
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
46
47
48
49
50
51
52
53
54
55
56
57
58
59
class AbstractStaticPageSeo(TimestampMixin, SeoMixin):
    """Store SEO metadata for named static pages with automatic cache invalidation.

    Saving the model invalidates the Redis entry managed by
    :func:`codex_django.core.redis.managers.seo.get_seo_redis_manager`.

    Admin:
        list_display:
            ("page_key", "seo_title", "updated_at")
        search_fields:
            ("page_key", "seo_title", "seo_description")
        readonly_fields:
            ("created_at", "updated_at")
    """

    page_key = models.CharField(
        max_length=50,
        unique=True,
        verbose_name=_("Page key"),
    )

    class Meta:
        abstract = True

    def __str__(self) -> str:
        """Return a readable admin label for the SEO record."""
        return f"SEO: {self.page_key}"

    def save(self, *args: Any, **kwargs: Any) -> None:
        """Persist the SEO record and invalidate the cached page payload.

        Args:
            *args: Positional arguments forwarded to ``models.Model.save()``.
            **kwargs: Keyword arguments forwarded to ``models.Model.save()``.
        """
        super().save(*args, **kwargs)
        # Invalidate SEO cache using manager
        manager = get_seo_redis_manager()
        manager.invalidate_page(self.page_key)
Functions
__str__()

Return a readable admin label for the SEO record.

Source code in src/codex_django/system/mixins/seo.py 
45
46
47
def __str__(self) -> str:
    """Return a readable admin label for the SEO record."""
    return f"SEO: {self.page_key}"
save(*args, **kwargs)

Persist the SEO record and invalidate the cached page payload.

Parameters:

Name Type Description Default
*args Any

Positional arguments forwarded to models.Model.save().

 ()
**kwargs Any

Keyword arguments forwarded to models.Model.save().

 {}
Source code in src/codex_django/system/mixins/seo.py 
49
50
51
52
53
54
55
56
57
58
59
def save(self, *args: Any, **kwargs: Any) -> None:
    """Persist the SEO record and invalidate the cached page payload.

    Args:
        *args: Positional arguments forwarded to ``models.Model.save()``.
        **kwargs: Keyword arguments forwarded to ``models.Model.save()``.
    """
    super().save(*args, **kwargs)
    # Invalidate SEO cache using manager
    manager = get_seo_redis_manager()
    manager.invalidate_page(self.page_key)

Functions

Translation mixins

codex_django.system.mixins.translations

Reusable model mixin for editable static translations.

Examples:

Keep simple key-value content in a project model::

from codex_django.system.mixins.translations import AbstractStaticTranslation

class StaticTranslation(AbstractStaticTranslation):
    pass

Classes

AbstractStaticTranslation

Bases: TimestampMixin

Store editable key-value content fragments managed through Django admin.

This mixin is useful for lightweight static copy that should be editable without introducing a more complex CMS or translation workflow.

Admin

list_display: ("key", "updated_at") search_fields: ("key", "content") readonly_fields: ("created_at", "updated_at")

Source code in src/codex_django/system/mixins/translations.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
46
47
48
49
50
51
52
class AbstractStaticTranslation(TimestampMixin):
    """Store editable key-value content fragments managed through Django admin.

    This mixin is useful for lightweight static copy that should be editable
    without introducing a more complex CMS or translation workflow.

    Admin:
        list_display:
            ("key", "updated_at")
        search_fields:
            ("key", "content")
        readonly_fields:
            ("created_at", "updated_at")
    """

    key = models.CharField(
        _("Key"),
        max_length=255,
        unique=True,
        help_text=_("Unique identifier for the content piece."),
    )
    content = models.TextField(
        _("Content"),
        blank=True,
        help_text=_("The translated content."),
    )

    class Meta:
        abstract = True
        verbose_name = _("Static Translation")
        verbose_name_plural = _("Static Translations")

    def __str__(self) -> str:
        """Return the translation key for admin and debug output."""
        return self.key
Functions
__str__()

Return the translation key for admin and debug output.

Source code in src/codex_django/system/mixins/translations.py 
50
51
52
def __str__(self) -> str:
    """Return the translation key for admin and debug output."""
    return self.key

User profile mixins

codex_django.system.mixins.user_profile

Reusable abstract mixin for a project-specific user profile model.

The target project typically subclasses this model in its own concrete UserProfile implementation, often generated by the cabinet/client scaffolding flows.

Examples:

Extend the profile with project-specific fields::

from codex_django.system.mixins.user_profile import AbstractUserProfile

class UserProfile(AbstractUserProfile):
    loyalty_tier = models.CharField(max_length=32, blank=True)

Classes

AbstractUserProfile

Bases: Model

Reusable abstract profile model linked one-to-one with the auth user.

The scaffolded project can subclass this mixin to keep personal data, acquisition metadata, and presentation helpers in a single profile model.

Admin

list_display: ("id", "user", "last_name", "first_name", "phone", "source", "created_at") search_fields: ("user__username", "user__email", "first_name", "last_name", "phone") list_filter: ("source", "created_at") readonly_fields: ("created_at", "updated_at")

Source code in src/codex_django/system/mixins/user_profile.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
class AbstractUserProfile(models.Model):
    """Reusable abstract profile model linked one-to-one with the auth user.

    The scaffolded project can subclass this mixin to keep personal data,
    acquisition metadata, and presentation helpers in a single profile model.

    Admin:
        list_display:
            ("id", "user", "last_name", "first_name", "phone", "source", "created_at")
        search_fields:
            ("user__username", "user__email", "first_name", "last_name", "phone")
        list_filter:
            ("source", "created_at")
        readonly_fields:
            ("created_at", "updated_at")
    """

    user = models.OneToOneField(
        settings.AUTH_USER_MODEL,
        on_delete=models.CASCADE,
        null=True,
        blank=True,
        related_name="profile",
    )

    # ── Personal data ──────────────────────────────────────────────
    first_name = models.CharField(max_length=100, blank=True)
    last_name = models.CharField(max_length=100, blank=True)
    patronymic = models.CharField(max_length=100, blank=True)
    phone = models.CharField(max_length=20, blank=True)
    birth_date = models.DateField(null=True, blank=True)
    avatar = models.ImageField(upload_to="avatars/", null=True, blank=True)

    # ── Acquisition ────────────────────────────────────────────────
    source = models.CharField(
        max_length=50,
        blank=True,
        help_text="How the profile was created: booking / form / manual / allauth",
    )
    notes = models.TextField(blank=True)

    # ── Meta timestamps ────────────────────────────────────────────
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    class Meta:
        abstract = True

    # ── Helpers ─────────────────────────────────────────────────────

    def get_full_name(self) -> str:
        """Return the profile's full name in ``Last First Patronymic`` order.

        Returns:
            A whitespace-normalized full name string assembled from the
            available name parts.
        """
        parts = [self.last_name, self.first_name, self.patronymic]
        return " ".join(p for p in parts if p).strip()

    def get_initials(self) -> str:
        """Return up to two uppercase initials built from first and last name.

        Returns:
            A two-letter initials string when name parts exist, otherwise ``?``.
        """
        initials = ""
        if self.first_name:
            initials += self.first_name[0].upper()
        if self.last_name:
            initials += self.last_name[0].upper()
        return initials or "?"
Functions
get_full_name()

Return the profile's full name in Last First Patronymic order.

Returns:

Type Description
str

A whitespace-normalized full name string assembled from the
str

available name parts.
Source code in src/codex_django/system/mixins/user_profile.py 
72
73
74
75
76
77
78
79
80
def get_full_name(self) -> str:
    """Return the profile's full name in ``Last First Patronymic`` order.

    Returns:
        A whitespace-normalized full name string assembled from the
        available name parts.
    """
    parts = [self.last_name, self.first_name, self.patronymic]
    return " ".join(p for p in parts if p).strip()
get_initials()

Return up to two uppercase initials built from first and last name.

Returns:

Type Description
str

A two-letter initials string when name parts exist, otherwise ?.
Source code in src/codex_django/system/mixins/user_profile.py 
82
83
84
85
86
87
88
89
90
91
92
93
def get_initials(self) -> str:
    """Return up to two uppercase initials built from first and last name.

    Returns:
        A two-letter initials string when name parts exist, otherwise ``?``.
    """
    initials = ""
    if self.first_name:
        initials += self.first_name[0].upper()
    if self.last_name:
        initials += self.last_name[0].upper()
    return initials or "?"

Integration mixins

codex_django.system.mixins.integrations

Integration mixins for storing provider credentials in site settings models.

Compose these mixins with :class:codex_django.system.mixins.settings.AbstractSiteSettings to keep third-party credentials in a single singleton-like settings object. Each mixin documents the Django admin wiring that should be added when the fields are exposed in the project's ModelAdmin.

Classes

GoogleIntegrationsMixin

Bases: Model

Add Google Maps, Business, and reCAPTCHA credentials.

Notes

Secret values that should not be exposed publicly are stored in encrypted fields where appropriate.

Admin

fieldsets: ( _("Google Services"), { "fields": ( "google_maps_api_key", "google_business_api_key", "google_recaptcha_site_key", "google_recaptcha_secret_key", ), "classes": ("collapse",), }, )

Source code in src/codex_django/system/mixins/integrations.py 
14
15
16
17
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
class GoogleIntegrationsMixin(models.Model):
    """Add Google Maps, Business, and reCAPTCHA credentials.

    Notes:
        Secret values that should not be exposed publicly are stored in
        encrypted fields where appropriate.

    Admin:
        fieldsets: (
            _("Google Services"),
            {
                "fields": (
                    "google_maps_api_key",
                    "google_business_api_key",
                    "google_recaptcha_site_key",
                    "google_recaptcha_secret_key",
                ),
                "classes": ("collapse",),
            },
        )
    """

    google_maps_api_key = models.CharField(_("Google Maps API Key"), max_length=255, blank=True)
    google_business_api_key = models.CharField(_("Google Business API Key"), max_length=255, blank=True)
    google_recaptcha_site_key = models.CharField(_("reCAPTCHA Site Key"), max_length=255, blank=True)
    google_recaptcha_secret_key = EncryptedCharField(_("reCAPTCHA Secret Key"), max_length=255, blank=True)

    class Meta:
        abstract = True

MetaIntegrationsMixin

Bases: Model

Add Meta platform credentials for Facebook and Instagram integrations.

Notes

This mixin is aimed at ad pixels and app-level integrations rather than content embeds.

Admin

fieldsets: ( _("Meta (Facebook/Instagram)"), { "fields": ("facebook_pixel_id", "facebook_app_id", "facebook_app_secret"), "classes": ("collapse",), }, )

Source code in src/codex_django/system/mixins/integrations.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
class MetaIntegrationsMixin(models.Model):
    """Add Meta platform credentials for Facebook and Instagram integrations.

    Notes:
        This mixin is aimed at ad pixels and app-level integrations rather
        than content embeds.

    Admin:
        fieldsets: (
            _("Meta (Facebook/Instagram)"),
            {
                "fields": ("facebook_pixel_id", "facebook_app_id", "facebook_app_secret"),
                "classes": ("collapse",),
            },
        )
    """

    facebook_pixel_id = models.CharField(_("Facebook Pixel ID"), max_length=255, blank=True)
    facebook_app_id = models.CharField(_("Facebook App ID"), max_length=255, blank=True)
    facebook_app_secret = EncryptedCharField(_("Facebook App Secret"), max_length=255, blank=True)

    class Meta:
        abstract = True

StripeIntegrationsMixin

Bases: Model

Add Stripe payment credentials and webhook secrets.

Notes

The secret and webhook fields use encrypted storage because they are server-side credentials.

Admin

fieldsets: ( _("Stripe Payments"), { "fields": ("stripe_public_key", "stripe_secret_key", "stripe_webhook_secret"), "classes": ("collapse",), }, )

Source code in src/codex_django/system/mixins/integrations.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
class StripeIntegrationsMixin(models.Model):
    """Add Stripe payment credentials and webhook secrets.

    Notes:
        The secret and webhook fields use encrypted storage because they are
        server-side credentials.

    Admin:
        fieldsets: (
            _("Stripe Payments"),
            {
                "fields": ("stripe_public_key", "stripe_secret_key", "stripe_webhook_secret"),
                "classes": ("collapse",),
            },
        )
    """

    stripe_public_key = models.CharField(_("Stripe Public Key"), max_length=255, blank=True)
    stripe_secret_key = EncryptedCharField(_("Stripe Secret Key"), max_length=255, blank=True)
    stripe_webhook_secret = EncryptedCharField(_("Stripe Webhook Secret"), max_length=255, blank=True)

    class Meta:
        abstract = True

CrmIntegrationsMixin

Bases: Model

Add a generic CRM endpoint and API key pair.

Notes

This mixin intentionally stays provider-agnostic so projects can wire custom CRM clients around a simple URL + credential contract.

Admin

fieldsets: (_("CRM Integration"), {"fields": ("crm_api_url", "crm_api_key"), "classes": ("collapse",)})

Source code in src/codex_django/system/mixins/integrations.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
class CrmIntegrationsMixin(models.Model):
    """Add a generic CRM endpoint and API key pair.

    Notes:
        This mixin intentionally stays provider-agnostic so projects can wire
        custom CRM clients around a simple URL + credential contract.

    Admin:
        fieldsets:
            (_("CRM Integration"), {"fields": ("crm_api_url", "crm_api_key"), "classes": ("collapse",)})
    """

    crm_api_url = models.URLField(_("CRM API URL"), blank=True)
    crm_api_key = EncryptedCharField(_("CRM API Key"), max_length=255, blank=True)

    class Meta:
        abstract = True

TwilioIntegrationsMixin

Bases: Model

Add Twilio credentials for SMS and WhatsApp delivery.

Notes

The mixin stores both SMS and WhatsApp sender configuration because projects often enable them together.

Admin

fieldsets: ( _("Twilio (SMS / WhatsApp)"), { "fields": ( "twilio_account_sid", "twilio_auth_token", "twilio_from_number", "twilio_whatsapp_from", ), "classes": ("collapse",), }, )

Source code in src/codex_django/system/mixins/integrations.py 
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
class TwilioIntegrationsMixin(models.Model):
    """Add Twilio credentials for SMS and WhatsApp delivery.

    Notes:
        The mixin stores both SMS and WhatsApp sender configuration because
        projects often enable them together.

    Admin:
        fieldsets: (
            _("Twilio (SMS / WhatsApp)"),
            {
                "fields": (
                    "twilio_account_sid",
                    "twilio_auth_token",
                    "twilio_from_number",
                    "twilio_whatsapp_from",
                ),
                "classes": ("collapse",),
            },
        )
    """

    twilio_account_sid = models.CharField(_("Twilio Account SID"), max_length=255, blank=True)
    twilio_auth_token = EncryptedCharField(_("Twilio Auth Token"), max_length=255, blank=True)
    twilio_from_number = models.CharField(
        _("Twilio From Number"), max_length=50, blank=True, help_text=_("E.g. +15017122661")
    )
    twilio_whatsapp_from = models.CharField(
        _("Twilio WhatsApp From"), max_length=50, blank=True, help_text=_("E.g. +14155238886")
    )

    class Meta:
        abstract = True

SevenIoIntegrationsMixin

Bases: Model

Add Seven.io credentials for SMS and WhatsApp delivery.

Notes

Use this mixin as a lightweight alternative to Twilio when Seven.io is the preferred outbound provider.

Admin

fieldsets: (_("Seven.io (SMS / WhatsApp)"), {"fields": ("seven_api_key", "seven_sender_id"), "classes": ("collapse",)})

Source code in src/codex_django/system/mixins/integrations.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
class SevenIoIntegrationsMixin(models.Model):
    """Add Seven.io credentials for SMS and WhatsApp delivery.

    Notes:
        Use this mixin as a lightweight alternative to Twilio when Seven.io
        is the preferred outbound provider.

    Admin:
        fieldsets:
            (_("Seven.io (SMS / WhatsApp)"), {"fields": ("seven_api_key", "seven_sender_id"), "classes": ("collapse",)})
    """

    seven_api_key = EncryptedCharField(_("Seven.io API Key"), max_length=255, blank=True)
    seven_sender_id = models.CharField(
        _("Seven.io Sender ID"), max_length=50, blank=True, help_text=_("SMS sender name or number")
    )

    class Meta:
        abstract = True

ExtraIntegrationsMixin

Bases: Model

Add a JSON field for custom or project-specific integration settings.

Notes

This field is best used as an extension point for values that do not justify a dedicated strongly typed mixin yet.

Admin

fieldsets: (_("Extra Integrations"), {"fields": ("extra_config",), "classes": ("collapse",)})

Source code in src/codex_django/system/mixins/integrations.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
class ExtraIntegrationsMixin(models.Model):
    """Add a JSON field for custom or project-specific integration settings.

    Notes:
        This field is best used as an extension point for values that do not
        justify a dedicated strongly typed mixin yet.

    Admin:
        fieldsets:
            (_("Extra Integrations"), {"fields": ("extra_config",), "classes": ("collapse",)})
    """

    extra_config = models.JSONField(_("Extra Configuration"), default=dict, blank=True)

    class Meta:
        abstract = True

Base management commands

codex_django.system.management.base_commands

Base classes and helpers for content update and fixture import commands.

These command abstractions encapsulate two recurring patterns:

  • running several update commands as one orchestration step
  • skipping fixture imports when the input files have not changed
  • importing Django-style JSON fixture payloads into configurable models

Classes

JsonFixtureLoadResult dataclass

Loaded Django-style fixture rows plus parsing diagnostics.

Source code in src/codex_django/system/management/base_commands.py 
26
27
28
29
30
31
32
33
34
35
36
37
@dataclass(frozen=True)
class JsonFixtureLoadResult:
    """Loaded Django-style fixture rows plus parsing diagnostics."""

    rows: list[dict[str, Any]] = field(default_factory=list)
    skipped: int = 0
    errors: list[str] = field(default_factory=list)

    @property
    def success(self) -> bool:
        """Return whether the fixture file was parsed successfully."""
        return not self.errors
Attributes
success property

Return whether the fixture file was parsed successfully.

FixtureImportResult dataclass

Summary of model rows created, updated, or skipped by an importer.

Source code in src/codex_django/system/management/base_commands.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
@dataclass(frozen=True)
class FixtureImportResult:
    """Summary of model rows created, updated, or skipped by an importer."""

    created: int = 0
    updated: int = 0
    skipped: int = 0
    errors: list[str] = field(default_factory=list)

    @property
    def success(self) -> bool:
        """Return whether the import completed without fatal errors."""
        return not self.errors
Attributes
success property

Return whether the import completed without fatal errors.

BaseUpdateAllContentCommand

Bases: BaseCommand

Run a predefined sequence of content update subcommands.

Subclasses are expected to populate commands_to_run with Django management command names. Each subcommand receives the --force flag when it is provided to the aggregate command.

Source code in src/codex_django/system/management/base_commands.py 
 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
class BaseUpdateAllContentCommand(BaseCommand):
    """Run a predefined sequence of content update subcommands.

    Subclasses are expected to populate ``commands_to_run`` with Django
    management command names. Each subcommand receives the ``--force`` flag
    when it is provided to the aggregate command.
    """

    help = "Run multiple content update commands"
    commands_to_run: ClassVar[list[str]] = []

    def add_arguments(self, parser: ArgumentParser) -> None:
        """Register shared command-line arguments.

        Args:
            parser: Django/argparse parser instance for this command.
        """
        parser.add_argument("--force", action="store_true", help="Ignore hash checks and force all updates")

    def get_command_label(self, command_name: str) -> str:
        """Return a display label for a subcommand.

        Args:
            command_name: Django management command name.

        Returns:
            Human-readable label. Subclasses may override this to print
            project-specific section headings.
        """
        return command_name

    def before_subcommand(self, command_name: str) -> None:
        """Hook called before each subcommand runs."""

    def after_subcommand(self, command_name: str) -> None:
        """Hook called after each successful subcommand runs."""

    def handle(self, *args: Any, **options: Any) -> None:
        """Run each configured subcommand and aggregate failures.

        Args:
            *args: Positional arguments forwarded by Django's command runner.
            **options: Parsed command-line options.

        Raises:
            django.core.management.base.CommandError: If one or more
                subcommands fail.
        """
        log.info(f"Command: {self.__class__.__module__.split('.')[-1]} | Action: Start")
        force = options.get("force", False)
        errors = []

        if not self.commands_to_run:
            self.stdout.write(self.style.WARNING("No commands defined in `commands_to_run`."))
            return

        for cmd in self.commands_to_run:
            try:
                label = self.get_command_label(cmd)
                log.debug(f"Action: SubCommand | name={cmd} label={label}")
                self.before_subcommand(cmd)
                log.debug(f"Action: SubCommand | name={cmd}")
                call_command(cmd, force=force)
                self.after_subcommand(cmd)
            except Exception as e:
                msg = f"Subcommand {cmd} failed: {e}"
                log.error(f"Action: SubError | error={msg}")
                errors.append(msg)

        if errors:
            self.stdout.write(self.style.ERROR(f"Update completed with {len(errors)} errors."))
            for err in errors:
                self.stdout.write(self.style.ERROR(f" - {err}"))
            raise CommandError("One or more subcommands failed.")

        log.info("Action: Success")
        self.stdout.write(self.style.SUCCESS("\nAll updates completed. Cache invalidated selectively per change."))
Functions
add_arguments(parser)

Register shared command-line arguments.

Parameters:

Name Type Description Default
parser ArgumentParser

Django/argparse parser instance for this command.

 required
Source code in src/codex_django/system/management/base_commands.py 
104
105
106
107
108
109
110
def add_arguments(self, parser: ArgumentParser) -> None:
    """Register shared command-line arguments.

    Args:
        parser: Django/argparse parser instance for this command.
    """
    parser.add_argument("--force", action="store_true", help="Ignore hash checks and force all updates")
get_command_label(command_name)

Return a display label for a subcommand.

Parameters:

Name Type Description Default
command_name str

Django management command name.

 required

Returns:

Type Description
str

Human-readable label. Subclasses may override this to print
str

project-specific section headings.
Source code in src/codex_django/system/management/base_commands.py 
112
113
114
115
116
117
118
119
120
121
122
def get_command_label(self, command_name: str) -> str:
    """Return a display label for a subcommand.

    Args:
        command_name: Django management command name.

    Returns:
        Human-readable label. Subclasses may override this to print
        project-specific section headings.
    """
    return command_name
before_subcommand(command_name)

Hook called before each subcommand runs.

Source code in src/codex_django/system/management/base_commands.py 
124
125
def before_subcommand(self, command_name: str) -> None:
    """Hook called before each subcommand runs."""
after_subcommand(command_name)

Hook called after each successful subcommand runs.

Source code in src/codex_django/system/management/base_commands.py 
127
128
def after_subcommand(self, command_name: str) -> None:
    """Hook called after each successful subcommand runs."""
handle(*args, **options)

Run each configured subcommand and aggregate failures.

Parameters:

Name Type Description Default
*args Any

Positional arguments forwarded by Django's command runner.

 ()
**options Any

Parsed command-line options.

 {}

Raises:

Type Description
CommandError

If one or more subcommands fail.
Source code in src/codex_django/system/management/base_commands.py 
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
def handle(self, *args: Any, **options: Any) -> None:
    """Run each configured subcommand and aggregate failures.

    Args:
        *args: Positional arguments forwarded by Django's command runner.
        **options: Parsed command-line options.

    Raises:
        django.core.management.base.CommandError: If one or more
            subcommands fail.
    """
    log.info(f"Command: {self.__class__.__module__.split('.')[-1]} | Action: Start")
    force = options.get("force", False)
    errors = []

    if not self.commands_to_run:
        self.stdout.write(self.style.WARNING("No commands defined in `commands_to_run`."))
        return

    for cmd in self.commands_to_run:
        try:
            label = self.get_command_label(cmd)
            log.debug(f"Action: SubCommand | name={cmd} label={label}")
            self.before_subcommand(cmd)
            log.debug(f"Action: SubCommand | name={cmd}")
            call_command(cmd, force=force)
            self.after_subcommand(cmd)
        except Exception as e:
            msg = f"Subcommand {cmd} failed: {e}"
            log.error(f"Action: SubError | error={msg}")
            errors.append(msg)

    if errors:
        self.stdout.write(self.style.ERROR(f"Update completed with {len(errors)} errors."))
        for err in errors:
            self.stdout.write(self.style.ERROR(f" - {err}"))
        raise CommandError("One or more subcommands failed.")

    log.info("Action: Success")
    self.stdout.write(self.style.SUCCESS("\nAll updates completed. Cache invalidated selectively per change."))

BaseHashProtectedCommand

Bases: BaseCommand

Base class for fixture import commands guarded by a Redis hash check.

The command computes a combined hash for the declared fixture files and skips the import when the stored hash matches the current one, unless the caller passes --force.

Source code in src/codex_django/system/management/base_commands.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
class BaseHashProtectedCommand(BaseCommand):
    """Base class for fixture import commands guarded by a Redis hash check.

    The command computes a combined hash for the declared fixture files and
    skips the import when the stored hash matches the current one, unless the
    caller passes ``--force``.
    """

    fixture_key: str = ""

    def add_arguments(self, parser: ArgumentParser) -> None:
        """Register shared command-line arguments.

        Args:
            parser: Django/argparse parser instance for this command.
        """
        parser.add_argument("--force", action="store_true", help="Ignore hash check and force update")

    def get_fixture_paths(self) -> list[Path]:
        """Return the fixture files that define the current content payload.

        Returns:
            A list of filesystem paths that should participate in the content
            hash calculation.
        """
        raise NotImplementedError("Subclasses must implement get_fixture_paths().")

    def handle_import(self, *args: Any, **options: Any) -> bool:
        """Execute the actual import operation for the selected fixtures.

        Args:
            *args: Positional arguments forwarded by Django's management
                command infrastructure.
            **options: Parsed command-line options.

        Returns:
            ``True`` when the import completed successfully. Returning
            ``False`` prevents the stored fixture hash from being updated.
        """
        raise NotImplementedError("Subclasses must implement handle_import().")

    def handle(self, *args: Any, **options: Any) -> None:
        """Skip or execute an import based on the computed fixture hash.

        Args:
            *args: Positional arguments forwarded by Django's command runner.
            **options: Parsed command-line options.

        Raises:
            ValueError: If the subclass did not set ``fixture_key``.
        """
        if not self.fixture_key:
            raise ValueError("fixture_key is not set on the command class")

        paths = [p for p in self.get_fixture_paths() if p.is_file()]
        if not paths:
            self.stdout.write(self.style.WARNING(f"Skipped {self.fixture_key}: No fixtures found."))
            return

        current_hash = compute_paths_hash(paths)
        force = options.get("force", False)

        manager = get_fixture_hash_manager()
        stored_hash = manager.get_hash(self.fixture_key)

        if not force and stored_hash == current_hash:
            self.stdout.write(f"Skipped {self.fixture_key}: fixture hash unchanged.")
            return

        # Execute the actual import logic
        success = self.handle_import(*args, **options)

        # Update hash if import succeeded (or if it doesn't return anything)
        if success is not False:
            manager.set_hash(self.fixture_key, current_hash)
            log.debug(f"Action: UpdateHash | fixture={self.fixture_key} hash={current_hash}")
Functions
add_arguments(parser)

Register shared command-line arguments.

Parameters:

Name Type Description Default
parser ArgumentParser

Django/argparse parser instance for this command.

 required
Source code in src/codex_django/system/management/base_commands.py 
182
183
184
185
186
187
188
def add_arguments(self, parser: ArgumentParser) -> None:
    """Register shared command-line arguments.

    Args:
        parser: Django/argparse parser instance for this command.
    """
    parser.add_argument("--force", action="store_true", help="Ignore hash check and force update")
get_fixture_paths()

Return the fixture files that define the current content payload.

Returns:

Type Description
list[Path]

A list of filesystem paths that should participate in the content
list[Path]

hash calculation.
Source code in src/codex_django/system/management/base_commands.py 
190
191
192
193
194
195
196
197
def get_fixture_paths(self) -> list[Path]:
    """Return the fixture files that define the current content payload.

    Returns:
        A list of filesystem paths that should participate in the content
        hash calculation.
    """
    raise NotImplementedError("Subclasses must implement get_fixture_paths().")
handle_import(*args, **options)

Execute the actual import operation for the selected fixtures.

Parameters:

Name Type Description Default
*args Any

Positional arguments forwarded by Django's management command infrastructure.

 ()
**options Any

Parsed command-line options.

 {}

Returns:

Type Description
bool

True when the import completed successfully. Returning
bool

False prevents the stored fixture hash from being updated.
Source code in src/codex_django/system/management/base_commands.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
def handle_import(self, *args: Any, **options: Any) -> bool:
    """Execute the actual import operation for the selected fixtures.

    Args:
        *args: Positional arguments forwarded by Django's management
            command infrastructure.
        **options: Parsed command-line options.

    Returns:
        ``True`` when the import completed successfully. Returning
        ``False`` prevents the stored fixture hash from being updated.
    """
    raise NotImplementedError("Subclasses must implement handle_import().")
handle(*args, **options)

Skip or execute an import based on the computed fixture hash.

Parameters:

Name Type Description Default
*args Any

Positional arguments forwarded by Django's command runner.

 ()
**options Any

Parsed command-line options.

 {}

Raises:

Type Description
ValueError

If the subclass did not set fixture_key.
Source code in src/codex_django/system/management/base_commands.py 
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
def handle(self, *args: Any, **options: Any) -> None:
    """Skip or execute an import based on the computed fixture hash.

    Args:
        *args: Positional arguments forwarded by Django's command runner.
        **options: Parsed command-line options.

    Raises:
        ValueError: If the subclass did not set ``fixture_key``.
    """
    if not self.fixture_key:
        raise ValueError("fixture_key is not set on the command class")

    paths = [p for p in self.get_fixture_paths() if p.is_file()]
    if not paths:
        self.stdout.write(self.style.WARNING(f"Skipped {self.fixture_key}: No fixtures found."))
        return

    current_hash = compute_paths_hash(paths)
    force = options.get("force", False)

    manager = get_fixture_hash_manager()
    stored_hash = manager.get_hash(self.fixture_key)

    if not force and stored_hash == current_hash:
        self.stdout.write(f"Skipped {self.fixture_key}: fixture hash unchanged.")
        return

    # Execute the actual import logic
    success = self.handle_import(*args, **options)

    # Update hash if import succeeded (or if it doesn't return anything)
    if success is not False:
        manager.set_hash(self.fixture_key, current_hash)
        log.debug(f"Action: UpdateHash | fixture={self.fixture_key} hash={current_hash}")

JsonFixtureUpsertCommand

Bases: BaseHashProtectedCommand

Import Django-style JSON fixture rows with update_or_create.

Subclasses usually set fixture_path, fixture_key, model_path or model_class, and lookup_field.

Source code in src/codex_django/system/management/base_commands.py 
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
class JsonFixtureUpsertCommand(BaseHashProtectedCommand):
    """Import Django-style JSON fixture rows with ``update_or_create``.

    Subclasses usually set ``fixture_path``, ``fixture_key``, ``model_path``
    or ``model_class``, and ``lookup_field``.
    """

    fixture_path: str | Path | None = None
    fixture_paths: ClassVar[list[str | Path]] = []
    model_path: str = ""
    model_class: Any = None
    lookup_field: str = "key"
    fields_key: str = "fields"

    def get_fixture_paths(self) -> list[Path]:
        """Return configured fixture paths as :class:`Path` objects."""
        configured = list(self.fixture_paths)
        if self.fixture_path is not None:
            configured.append(self.fixture_path)
        return [Path(path) for path in configured]

    def get_model_class(self) -> Any:
        """Return the model class that receives imported rows."""
        if self.model_class is not None:
            return self.model_class
        if not self.model_path:
            raise ValueError("model_path or model_class is required")
        return apps.get_model(self.model_path)

    def get_lookup_value(self, fields: dict[str, Any]) -> Any:
        """Return the lookup value for a fixture row."""
        return fields.get(self.lookup_field)

    def get_defaults(self, fields: dict[str, Any]) -> dict[str, Any]:
        """Return defaults passed to ``update_or_create``."""
        return fields

    def handle_import(self, *args: Any, **options: Any) -> bool:
        """Import configured JSON fixture rows into the target model."""
        model = self.get_model_class()
        created_count = 0
        updated_count = 0
        skipped_count = 0
        errors: list[str] = []

        for fixture_path in self.get_fixture_paths():
            load_result = load_json_fixture_rows(fixture_path, fields_key=self.fields_key)
            skipped_count += load_result.skipped
            if not load_result.success:
                errors.extend(load_result.errors)
                continue

            for fields in load_result.rows:
                lookup_value = self.get_lookup_value(fields)
                if lookup_value in (None, ""):
                    skipped_count += 1
                    continue

                _, created = model.objects.update_or_create(
                    **{self.lookup_field: lookup_value},
                    defaults=self.get_defaults(fields),
                )
                if created:
                    created_count += 1
                else:
                    updated_count += 1

        result = FixtureImportResult(
            created=created_count,
            updated=updated_count,
            skipped=skipped_count,
            errors=errors,
        )
        self.import_result = result

        if not result.success:
            for error in result.errors:
                self.stdout.write(self.style.ERROR(error))
            return False

        self.stdout.write(
            self.style.SUCCESS(
                f"Imported {self.fixture_key}: {result.created} created, "
                f"{result.updated} updated, {result.skipped} skipped"
            )
        )
        return True
Functions
get_fixture_paths()

Return configured fixture paths as :class:Path objects.

Source code in src/codex_django/system/management/base_commands.py 
264
265
266
267
268
269
def get_fixture_paths(self) -> list[Path]:
    """Return configured fixture paths as :class:`Path` objects."""
    configured = list(self.fixture_paths)
    if self.fixture_path is not None:
        configured.append(self.fixture_path)
    return [Path(path) for path in configured]
get_model_class()

Return the model class that receives imported rows.

Source code in src/codex_django/system/management/base_commands.py 
271
272
273
274
275
276
277
def get_model_class(self) -> Any:
    """Return the model class that receives imported rows."""
    if self.model_class is not None:
        return self.model_class
    if not self.model_path:
        raise ValueError("model_path or model_class is required")
    return apps.get_model(self.model_path)
get_lookup_value(fields)

Return the lookup value for a fixture row.

Source code in src/codex_django/system/management/base_commands.py 
279
280
281
def get_lookup_value(self, fields: dict[str, Any]) -> Any:
    """Return the lookup value for a fixture row."""
    return fields.get(self.lookup_field)
get_defaults(fields)

Return defaults passed to update_or_create.

Source code in src/codex_django/system/management/base_commands.py 
283
284
285
def get_defaults(self, fields: dict[str, Any]) -> dict[str, Any]:
    """Return defaults passed to ``update_or_create``."""
    return fields
handle_import(*args, **options)

Import configured JSON fixture rows into the target model.

Source code in src/codex_django/system/management/base_commands.py 
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
def handle_import(self, *args: Any, **options: Any) -> bool:
    """Import configured JSON fixture rows into the target model."""
    model = self.get_model_class()
    created_count = 0
    updated_count = 0
    skipped_count = 0
    errors: list[str] = []

    for fixture_path in self.get_fixture_paths():
        load_result = load_json_fixture_rows(fixture_path, fields_key=self.fields_key)
        skipped_count += load_result.skipped
        if not load_result.success:
            errors.extend(load_result.errors)
            continue

        for fields in load_result.rows:
            lookup_value = self.get_lookup_value(fields)
            if lookup_value in (None, ""):
                skipped_count += 1
                continue

            _, created = model.objects.update_or_create(
                **{self.lookup_field: lookup_value},
                defaults=self.get_defaults(fields),
            )
            if created:
                created_count += 1
            else:
                updated_count += 1

    result = FixtureImportResult(
        created=created_count,
        updated=updated_count,
        skipped=skipped_count,
        errors=errors,
    )
    self.import_result = result

    if not result.success:
        for error in result.errors:
            self.stdout.write(self.style.ERROR(error))
        return False

    self.stdout.write(
        self.style.SUCCESS(
            f"Imported {self.fixture_key}: {result.created} created, "
            f"{result.updated} updated, {result.skipped} skipped"
        )
    )
    return True

SingletonFixtureUpdateCommand

Bases: BaseHashProtectedCommand

Update a singleton model instance from the first row in a JSON fixture.

Source code in src/codex_django/system/management/base_commands.py 
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
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
423
424
425
class SingletonFixtureUpdateCommand(BaseHashProtectedCommand):
    """Update a singleton model instance from the first row in a JSON fixture."""

    fixture_path: str | Path | None = None
    fixture_paths: ClassVar[list[str | Path]] = []
    model_path: str = ""
    model_class: Any = None
    singleton_pk: Any = 1
    fields_key: str = "fields"

    def get_fixture_paths(self) -> list[Path]:
        """Return configured fixture paths as :class:`Path` objects."""
        configured = list(self.fixture_paths)
        if self.fixture_path is not None:
            configured.append(self.fixture_path)
        return [Path(path) for path in configured]

    def get_model_class(self) -> Any:
        """Return the singleton model class."""
        if self.model_class is not None:
            return self.model_class
        if not self.model_path:
            raise ValueError("model_path or model_class is required")
        return apps.get_model(self.model_path)

    def get_singleton(self, model: Any) -> Any:
        """Return the singleton instance for the target model."""
        load = getattr(model, "load", None)
        if callable(load):
            return load()
        instance, _ = model.objects.get_or_create(pk=self.singleton_pk)
        return instance

    def normalize_current_value(self, field_name: str, value: Any) -> Any:
        """Normalize current model values before fixture comparison."""
        if hasattr(value, "__str__") and field_name in ["latitude", "longitude"]:
            return str(value)
        return value

    def get_update_fields(self, fixture_fields: dict[str, Any]) -> dict[str, Any]:
        """Return fixture fields eligible for singleton assignment."""
        return fixture_fields

    def sync_instance(self, instance: Any) -> None:
        """Synchronize the updated instance to Redis when a manager is available."""
        from codex_django.core.redis.managers.settings import get_site_settings_manager

        manager = get_site_settings_manager()
        manager.save_instance(instance)

    def handle_import(self, *args: Any, **options: Any) -> bool:
        """Apply the first fixture row to the configured singleton instance."""
        fixture_paths = self.get_fixture_paths()
        if not fixture_paths:
            self.stdout.write(self.style.ERROR(f"No fixture path configured for {self.fixture_key}."))
            return False

        load_result = load_json_fixture_rows(fixture_paths[0], fields_key=self.fields_key)
        if not load_result.success:
            for error in load_result.errors:
                self.stdout.write(self.style.ERROR(error))
            return False
        if not load_result.rows:
            self.stdout.write(self.style.ERROR(f"Invalid fixture format for {self.fixture_key}: no rows found."))
            return False

        model = self.get_model_class()
        instance = self.get_singleton(model)
        updated_fields: list[str] = []

        for field_name, new_value in self.get_update_fields(load_result.rows[0]).items():
            if not hasattr(instance, field_name):
                continue
            current_value = self.normalize_current_value(field_name, getattr(instance, field_name))
            if str(current_value) != str(new_value):
                setattr(instance, field_name, new_value)
                updated_fields.append(field_name)

        self.updated_fields = updated_fields
        if not updated_fields:
            self.stdout.write(self.style.SUCCESS(f"No changes needed for {self.fixture_key}."))
            return True

        instance.save()
        self.sync_instance(instance)
        self.stdout.write(self.style.SUCCESS(f"Updated {self.fixture_key}: {len(updated_fields)} fields changed"))
        return True
Functions
get_fixture_paths()

Return configured fixture paths as :class:Path objects.

Source code in src/codex_django/system/management/base_commands.py 
349
350
351
352
353
354
def get_fixture_paths(self) -> list[Path]:
    """Return configured fixture paths as :class:`Path` objects."""
    configured = list(self.fixture_paths)
    if self.fixture_path is not None:
        configured.append(self.fixture_path)
    return [Path(path) for path in configured]
get_model_class()

Return the singleton model class.

Source code in src/codex_django/system/management/base_commands.py 
356
357
358
359
360
361
362
def get_model_class(self) -> Any:
    """Return the singleton model class."""
    if self.model_class is not None:
        return self.model_class
    if not self.model_path:
        raise ValueError("model_path or model_class is required")
    return apps.get_model(self.model_path)
get_singleton(model)

Return the singleton instance for the target model.

Source code in src/codex_django/system/management/base_commands.py 
364
365
366
367
368
369
370
def get_singleton(self, model: Any) -> Any:
    """Return the singleton instance for the target model."""
    load = getattr(model, "load", None)
    if callable(load):
        return load()
    instance, _ = model.objects.get_or_create(pk=self.singleton_pk)
    return instance
normalize_current_value(field_name, value)

Normalize current model values before fixture comparison.

Source code in src/codex_django/system/management/base_commands.py 
372
373
374
375
376
def normalize_current_value(self, field_name: str, value: Any) -> Any:
    """Normalize current model values before fixture comparison."""
    if hasattr(value, "__str__") and field_name in ["latitude", "longitude"]:
        return str(value)
    return value
get_update_fields(fixture_fields)

Return fixture fields eligible for singleton assignment.

Source code in src/codex_django/system/management/base_commands.py 
378
379
380
def get_update_fields(self, fixture_fields: dict[str, Any]) -> dict[str, Any]:
    """Return fixture fields eligible for singleton assignment."""
    return fixture_fields
sync_instance(instance)

Synchronize the updated instance to Redis when a manager is available.

Source code in src/codex_django/system/management/base_commands.py 
382
383
384
385
386
387
def sync_instance(self, instance: Any) -> None:
    """Synchronize the updated instance to Redis when a manager is available."""
    from codex_django.core.redis.managers.settings import get_site_settings_manager

    manager = get_site_settings_manager()
    manager.save_instance(instance)
handle_import(*args, **options)

Apply the first fixture row to the configured singleton instance.

Source code in src/codex_django/system/management/base_commands.py 
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
423
424
425
def handle_import(self, *args: Any, **options: Any) -> bool:
    """Apply the first fixture row to the configured singleton instance."""
    fixture_paths = self.get_fixture_paths()
    if not fixture_paths:
        self.stdout.write(self.style.ERROR(f"No fixture path configured for {self.fixture_key}."))
        return False

    load_result = load_json_fixture_rows(fixture_paths[0], fields_key=self.fields_key)
    if not load_result.success:
        for error in load_result.errors:
            self.stdout.write(self.style.ERROR(error))
        return False
    if not load_result.rows:
        self.stdout.write(self.style.ERROR(f"Invalid fixture format for {self.fixture_key}: no rows found."))
        return False

    model = self.get_model_class()
    instance = self.get_singleton(model)
    updated_fields: list[str] = []

    for field_name, new_value in self.get_update_fields(load_result.rows[0]).items():
        if not hasattr(instance, field_name):
            continue
        current_value = self.normalize_current_value(field_name, getattr(instance, field_name))
        if str(current_value) != str(new_value):
            setattr(instance, field_name, new_value)
            updated_fields.append(field_name)

    self.updated_fields = updated_fields
    if not updated_fields:
        self.stdout.write(self.style.SUCCESS(f"No changes needed for {self.fixture_key}."))
        return True

    instance.save()
    self.sync_instance(instance)
    self.stdout.write(self.style.SUCCESS(f"Updated {self.fixture_key}: {len(updated_fields)} fields changed"))
    return True

Functions

load_json_fixture_rows(path, fields_key='fields')

Load a Django-style JSON fixture and return each item's fields mapping.

Parameters:

Name Type Description Default
path Path

Fixture file path.

 required
fields_key str

Item key containing model field values.

 'fields'

Returns:

Type Description
JsonFixtureLoadResult

Parsed row mappings plus skipped-item and error diagnostics. Invalid
JsonFixtureLoadResult

JSON, unreadable files, and non-list payloads are fatal errors. Items
JsonFixtureLoadResult

without a usable fields mapping are skipped.
Source code in src/codex_django/system/management/base_commands.py 
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
def load_json_fixture_rows(path: Path, fields_key: str = "fields") -> JsonFixtureLoadResult:
    """Load a Django-style JSON fixture and return each item's ``fields`` mapping.

    Args:
        path: Fixture file path.
        fields_key: Item key containing model field values.

    Returns:
        Parsed row mappings plus skipped-item and error diagnostics. Invalid
        JSON, unreadable files, and non-list payloads are fatal errors. Items
        without a usable fields mapping are skipped.
    """
    try:
        with open(path, encoding="utf-8") as f:
            payload = json.load(f)
    except json.JSONDecodeError as exc:
        return JsonFixtureLoadResult(errors=[f"Error decoding JSON fixture {path}: {exc}"])
    except OSError as exc:
        return JsonFixtureLoadResult(errors=[f"Error reading JSON fixture {path}: {exc}"])

    if not isinstance(payload, list):
        return JsonFixtureLoadResult(errors=[f"Invalid fixture format in {path}: expected a list."])

    rows: list[dict[str, Any]] = []
    skipped = 0
    for item in payload:
        if not isinstance(item, Mapping):
            skipped += 1
            continue
        fields_value = item.get(fields_key)
        if not isinstance(fields_value, Mapping):
            skipped += 1
            continue
        rows.append(dict(fields_value))

    return JsonFixtureLoadResult(rows=rows, skipped=skipped)

Public management exports

codex_django.system.management

Management-command infrastructure for codex-django system helpers.

Classes

BaseHashProtectedCommand

Bases: BaseCommand

Base class for fixture import commands guarded by a Redis hash check.

The command computes a combined hash for the declared fixture files and skips the import when the stored hash matches the current one, unless the caller passes --force.

Source code in src/codex_django/system/management/base_commands.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
class BaseHashProtectedCommand(BaseCommand):
    """Base class for fixture import commands guarded by a Redis hash check.

    The command computes a combined hash for the declared fixture files and
    skips the import when the stored hash matches the current one, unless the
    caller passes ``--force``.
    """

    fixture_key: str = ""

    def add_arguments(self, parser: ArgumentParser) -> None:
        """Register shared command-line arguments.

        Args:
            parser: Django/argparse parser instance for this command.
        """
        parser.add_argument("--force", action="store_true", help="Ignore hash check and force update")

    def get_fixture_paths(self) -> list[Path]:
        """Return the fixture files that define the current content payload.

        Returns:
            A list of filesystem paths that should participate in the content
            hash calculation.
        """
        raise NotImplementedError("Subclasses must implement get_fixture_paths().")

    def handle_import(self, *args: Any, **options: Any) -> bool:
        """Execute the actual import operation for the selected fixtures.

        Args:
            *args: Positional arguments forwarded by Django's management
                command infrastructure.
            **options: Parsed command-line options.

        Returns:
            ``True`` when the import completed successfully. Returning
            ``False`` prevents the stored fixture hash from being updated.
        """
        raise NotImplementedError("Subclasses must implement handle_import().")

    def handle(self, *args: Any, **options: Any) -> None:
        """Skip or execute an import based on the computed fixture hash.

        Args:
            *args: Positional arguments forwarded by Django's command runner.
            **options: Parsed command-line options.

        Raises:
            ValueError: If the subclass did not set ``fixture_key``.
        """
        if not self.fixture_key:
            raise ValueError("fixture_key is not set on the command class")

        paths = [p for p in self.get_fixture_paths() if p.is_file()]
        if not paths:
            self.stdout.write(self.style.WARNING(f"Skipped {self.fixture_key}: No fixtures found."))
            return

        current_hash = compute_paths_hash(paths)
        force = options.get("force", False)

        manager = get_fixture_hash_manager()
        stored_hash = manager.get_hash(self.fixture_key)

        if not force and stored_hash == current_hash:
            self.stdout.write(f"Skipped {self.fixture_key}: fixture hash unchanged.")
            return

        # Execute the actual import logic
        success = self.handle_import(*args, **options)

        # Update hash if import succeeded (or if it doesn't return anything)
        if success is not False:
            manager.set_hash(self.fixture_key, current_hash)
            log.debug(f"Action: UpdateHash | fixture={self.fixture_key} hash={current_hash}")
Functions
add_arguments(parser)

Register shared command-line arguments.

Parameters:

Name Type Description Default
parser ArgumentParser

Django/argparse parser instance for this command.

 required
Source code in src/codex_django/system/management/base_commands.py 
182
183
184
185
186
187
188
def add_arguments(self, parser: ArgumentParser) -> None:
    """Register shared command-line arguments.

    Args:
        parser: Django/argparse parser instance for this command.
    """
    parser.add_argument("--force", action="store_true", help="Ignore hash check and force update")
get_fixture_paths()

Return the fixture files that define the current content payload.

Returns:

Type Description
list[Path]

A list of filesystem paths that should participate in the content
list[Path]

hash calculation.
Source code in src/codex_django/system/management/base_commands.py 
190
191
192
193
194
195
196
197
def get_fixture_paths(self) -> list[Path]:
    """Return the fixture files that define the current content payload.

    Returns:
        A list of filesystem paths that should participate in the content
        hash calculation.
    """
    raise NotImplementedError("Subclasses must implement get_fixture_paths().")
handle_import(*args, **options)

Execute the actual import operation for the selected fixtures.

Parameters:

Name Type Description Default
*args Any

Positional arguments forwarded by Django's management command infrastructure.

 ()
**options Any

Parsed command-line options.

 {}

Returns:

Type Description
bool

True when the import completed successfully. Returning
bool

False prevents the stored fixture hash from being updated.
Source code in src/codex_django/system/management/base_commands.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
def handle_import(self, *args: Any, **options: Any) -> bool:
    """Execute the actual import operation for the selected fixtures.

    Args:
        *args: Positional arguments forwarded by Django's management
            command infrastructure.
        **options: Parsed command-line options.

    Returns:
        ``True`` when the import completed successfully. Returning
        ``False`` prevents the stored fixture hash from being updated.
    """
    raise NotImplementedError("Subclasses must implement handle_import().")
handle(*args, **options)

Skip or execute an import based on the computed fixture hash.

Parameters:

Name Type Description Default
*args Any

Positional arguments forwarded by Django's command runner.

 ()
**options Any

Parsed command-line options.

 {}

Raises:

Type Description
ValueError

If the subclass did not set fixture_key.
Source code in src/codex_django/system/management/base_commands.py 
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
def handle(self, *args: Any, **options: Any) -> None:
    """Skip or execute an import based on the computed fixture hash.

    Args:
        *args: Positional arguments forwarded by Django's command runner.
        **options: Parsed command-line options.

    Raises:
        ValueError: If the subclass did not set ``fixture_key``.
    """
    if not self.fixture_key:
        raise ValueError("fixture_key is not set on the command class")

    paths = [p for p in self.get_fixture_paths() if p.is_file()]
    if not paths:
        self.stdout.write(self.style.WARNING(f"Skipped {self.fixture_key}: No fixtures found."))
        return

    current_hash = compute_paths_hash(paths)
    force = options.get("force", False)

    manager = get_fixture_hash_manager()
    stored_hash = manager.get_hash(self.fixture_key)

    if not force and stored_hash == current_hash:
        self.stdout.write(f"Skipped {self.fixture_key}: fixture hash unchanged.")
        return

    # Execute the actual import logic
    success = self.handle_import(*args, **options)

    # Update hash if import succeeded (or if it doesn't return anything)
    if success is not False:
        manager.set_hash(self.fixture_key, current_hash)
        log.debug(f"Action: UpdateHash | fixture={self.fixture_key} hash={current_hash}")

BaseUpdateAllContentCommand

Bases: BaseCommand

Run a predefined sequence of content update subcommands.

Subclasses are expected to populate commands_to_run with Django management command names. Each subcommand receives the --force flag when it is provided to the aggregate command.

Source code in src/codex_django/system/management/base_commands.py 
 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
class BaseUpdateAllContentCommand(BaseCommand):
    """Run a predefined sequence of content update subcommands.

    Subclasses are expected to populate ``commands_to_run`` with Django
    management command names. Each subcommand receives the ``--force`` flag
    when it is provided to the aggregate command.
    """

    help = "Run multiple content update commands"
    commands_to_run: ClassVar[list[str]] = []

    def add_arguments(self, parser: ArgumentParser) -> None:
        """Register shared command-line arguments.

        Args:
            parser: Django/argparse parser instance for this command.
        """
        parser.add_argument("--force", action="store_true", help="Ignore hash checks and force all updates")

    def get_command_label(self, command_name: str) -> str:
        """Return a display label for a subcommand.

        Args:
            command_name: Django management command name.

        Returns:
            Human-readable label. Subclasses may override this to print
            project-specific section headings.
        """
        return command_name

    def before_subcommand(self, command_name: str) -> None:
        """Hook called before each subcommand runs."""

    def after_subcommand(self, command_name: str) -> None:
        """Hook called after each successful subcommand runs."""

    def handle(self, *args: Any, **options: Any) -> None:
        """Run each configured subcommand and aggregate failures.

        Args:
            *args: Positional arguments forwarded by Django's command runner.
            **options: Parsed command-line options.

        Raises:
            django.core.management.base.CommandError: If one or more
                subcommands fail.
        """
        log.info(f"Command: {self.__class__.__module__.split('.')[-1]} | Action: Start")
        force = options.get("force", False)
        errors = []

        if not self.commands_to_run:
            self.stdout.write(self.style.WARNING("No commands defined in `commands_to_run`."))
            return

        for cmd in self.commands_to_run:
            try:
                label = self.get_command_label(cmd)
                log.debug(f"Action: SubCommand | name={cmd} label={label}")
                self.before_subcommand(cmd)
                log.debug(f"Action: SubCommand | name={cmd}")
                call_command(cmd, force=force)
                self.after_subcommand(cmd)
            except Exception as e:
                msg = f"Subcommand {cmd} failed: {e}"
                log.error(f"Action: SubError | error={msg}")
                errors.append(msg)

        if errors:
            self.stdout.write(self.style.ERROR(f"Update completed with {len(errors)} errors."))
            for err in errors:
                self.stdout.write(self.style.ERROR(f" - {err}"))
            raise CommandError("One or more subcommands failed.")

        log.info("Action: Success")
        self.stdout.write(self.style.SUCCESS("\nAll updates completed. Cache invalidated selectively per change."))
Functions
add_arguments(parser)

Register shared command-line arguments.

Parameters:

Name Type Description Default
parser ArgumentParser

Django/argparse parser instance for this command.

 required
Source code in src/codex_django/system/management/base_commands.py 
104
105
106
107
108
109
110
def add_arguments(self, parser: ArgumentParser) -> None:
    """Register shared command-line arguments.

    Args:
        parser: Django/argparse parser instance for this command.
    """
    parser.add_argument("--force", action="store_true", help="Ignore hash checks and force all updates")
get_command_label(command_name)

Return a display label for a subcommand.

Parameters:

Name Type Description Default
command_name str

Django management command name.

 required

Returns:

Type Description
str

Human-readable label. Subclasses may override this to print
str

project-specific section headings.
Source code in src/codex_django/system/management/base_commands.py 
112
113
114
115
116
117
118
119
120
121
122
def get_command_label(self, command_name: str) -> str:
    """Return a display label for a subcommand.

    Args:
        command_name: Django management command name.

    Returns:
        Human-readable label. Subclasses may override this to print
        project-specific section headings.
    """
    return command_name
before_subcommand(command_name)

Hook called before each subcommand runs.

Source code in src/codex_django/system/management/base_commands.py 
124
125
def before_subcommand(self, command_name: str) -> None:
    """Hook called before each subcommand runs."""
after_subcommand(command_name)

Hook called after each successful subcommand runs.

Source code in src/codex_django/system/management/base_commands.py 
127
128
def after_subcommand(self, command_name: str) -> None:
    """Hook called after each successful subcommand runs."""
handle(*args, **options)

Run each configured subcommand and aggregate failures.

Parameters:

Name Type Description Default
*args Any

Positional arguments forwarded by Django's command runner.

 ()
**options Any

Parsed command-line options.

 {}

Raises:

Type Description
CommandError

If one or more subcommands fail.
Source code in src/codex_django/system/management/base_commands.py 
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
def handle(self, *args: Any, **options: Any) -> None:
    """Run each configured subcommand and aggregate failures.

    Args:
        *args: Positional arguments forwarded by Django's command runner.
        **options: Parsed command-line options.

    Raises:
        django.core.management.base.CommandError: If one or more
            subcommands fail.
    """
    log.info(f"Command: {self.__class__.__module__.split('.')[-1]} | Action: Start")
    force = options.get("force", False)
    errors = []

    if not self.commands_to_run:
        self.stdout.write(self.style.WARNING("No commands defined in `commands_to_run`."))
        return

    for cmd in self.commands_to_run:
        try:
            label = self.get_command_label(cmd)
            log.debug(f"Action: SubCommand | name={cmd} label={label}")
            self.before_subcommand(cmd)
            log.debug(f"Action: SubCommand | name={cmd}")
            call_command(cmd, force=force)
            self.after_subcommand(cmd)
        except Exception as e:
            msg = f"Subcommand {cmd} failed: {e}"
            log.error(f"Action: SubError | error={msg}")
            errors.append(msg)

    if errors:
        self.stdout.write(self.style.ERROR(f"Update completed with {len(errors)} errors."))
        for err in errors:
            self.stdout.write(self.style.ERROR(f" - {err}"))
        raise CommandError("One or more subcommands failed.")

    log.info("Action: Success")
    self.stdout.write(self.style.SUCCESS("\nAll updates completed. Cache invalidated selectively per change."))

FixtureImportResult dataclass

Summary of model rows created, updated, or skipped by an importer.

Source code in src/codex_django/system/management/base_commands.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
@dataclass(frozen=True)
class FixtureImportResult:
    """Summary of model rows created, updated, or skipped by an importer."""

    created: int = 0
    updated: int = 0
    skipped: int = 0
    errors: list[str] = field(default_factory=list)

    @property
    def success(self) -> bool:
        """Return whether the import completed without fatal errors."""
        return not self.errors
Attributes
success property

Return whether the import completed without fatal errors.

JsonFixtureLoadResult dataclass

Loaded Django-style fixture rows plus parsing diagnostics.

Source code in src/codex_django/system/management/base_commands.py 
26
27
28
29
30
31
32
33
34
35
36
37
@dataclass(frozen=True)
class JsonFixtureLoadResult:
    """Loaded Django-style fixture rows plus parsing diagnostics."""

    rows: list[dict[str, Any]] = field(default_factory=list)
    skipped: int = 0
    errors: list[str] = field(default_factory=list)

    @property
    def success(self) -> bool:
        """Return whether the fixture file was parsed successfully."""
        return not self.errors
Attributes
success property

Return whether the fixture file was parsed successfully.

JsonFixtureUpsertCommand

Bases: BaseHashProtectedCommand

Import Django-style JSON fixture rows with update_or_create.

Subclasses usually set fixture_path, fixture_key, model_path or model_class, and lookup_field.

Source code in src/codex_django/system/management/base_commands.py 
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
class JsonFixtureUpsertCommand(BaseHashProtectedCommand):
    """Import Django-style JSON fixture rows with ``update_or_create``.

    Subclasses usually set ``fixture_path``, ``fixture_key``, ``model_path``
    or ``model_class``, and ``lookup_field``.
    """

    fixture_path: str | Path | None = None
    fixture_paths: ClassVar[list[str | Path]] = []
    model_path: str = ""
    model_class: Any = None
    lookup_field: str = "key"
    fields_key: str = "fields"

    def get_fixture_paths(self) -> list[Path]:
        """Return configured fixture paths as :class:`Path` objects."""
        configured = list(self.fixture_paths)
        if self.fixture_path is not None:
            configured.append(self.fixture_path)
        return [Path(path) for path in configured]

    def get_model_class(self) -> Any:
        """Return the model class that receives imported rows."""
        if self.model_class is not None:
            return self.model_class
        if not self.model_path:
            raise ValueError("model_path or model_class is required")
        return apps.get_model(self.model_path)

    def get_lookup_value(self, fields: dict[str, Any]) -> Any:
        """Return the lookup value for a fixture row."""
        return fields.get(self.lookup_field)

    def get_defaults(self, fields: dict[str, Any]) -> dict[str, Any]:
        """Return defaults passed to ``update_or_create``."""
        return fields

    def handle_import(self, *args: Any, **options: Any) -> bool:
        """Import configured JSON fixture rows into the target model."""
        model = self.get_model_class()
        created_count = 0
        updated_count = 0
        skipped_count = 0
        errors: list[str] = []

        for fixture_path in self.get_fixture_paths():
            load_result = load_json_fixture_rows(fixture_path, fields_key=self.fields_key)
            skipped_count += load_result.skipped
            if not load_result.success:
                errors.extend(load_result.errors)
                continue

            for fields in load_result.rows:
                lookup_value = self.get_lookup_value(fields)
                if lookup_value in (None, ""):
                    skipped_count += 1
                    continue

                _, created = model.objects.update_or_create(
                    **{self.lookup_field: lookup_value},
                    defaults=self.get_defaults(fields),
                )
                if created:
                    created_count += 1
                else:
                    updated_count += 1

        result = FixtureImportResult(
            created=created_count,
            updated=updated_count,
            skipped=skipped_count,
            errors=errors,
        )
        self.import_result = result

        if not result.success:
            for error in result.errors:
                self.stdout.write(self.style.ERROR(error))
            return False

        self.stdout.write(
            self.style.SUCCESS(
                f"Imported {self.fixture_key}: {result.created} created, "
                f"{result.updated} updated, {result.skipped} skipped"
            )
        )
        return True
Functions
get_fixture_paths()

Return configured fixture paths as :class:Path objects.

Source code in src/codex_django/system/management/base_commands.py 
264
265
266
267
268
269
def get_fixture_paths(self) -> list[Path]:
    """Return configured fixture paths as :class:`Path` objects."""
    configured = list(self.fixture_paths)
    if self.fixture_path is not None:
        configured.append(self.fixture_path)
    return [Path(path) for path in configured]
get_model_class()

Return the model class that receives imported rows.

Source code in src/codex_django/system/management/base_commands.py 
271
272
273
274
275
276
277
def get_model_class(self) -> Any:
    """Return the model class that receives imported rows."""
    if self.model_class is not None:
        return self.model_class
    if not self.model_path:
        raise ValueError("model_path or model_class is required")
    return apps.get_model(self.model_path)
get_lookup_value(fields)

Return the lookup value for a fixture row.

Source code in src/codex_django/system/management/base_commands.py 
279
280
281
def get_lookup_value(self, fields: dict[str, Any]) -> Any:
    """Return the lookup value for a fixture row."""
    return fields.get(self.lookup_field)
get_defaults(fields)

Return defaults passed to update_or_create.

Source code in src/codex_django/system/management/base_commands.py 
283
284
285
def get_defaults(self, fields: dict[str, Any]) -> dict[str, Any]:
    """Return defaults passed to ``update_or_create``."""
    return fields
handle_import(*args, **options)

Import configured JSON fixture rows into the target model.

Source code in src/codex_django/system/management/base_commands.py 
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
def handle_import(self, *args: Any, **options: Any) -> bool:
    """Import configured JSON fixture rows into the target model."""
    model = self.get_model_class()
    created_count = 0
    updated_count = 0
    skipped_count = 0
    errors: list[str] = []

    for fixture_path in self.get_fixture_paths():
        load_result = load_json_fixture_rows(fixture_path, fields_key=self.fields_key)
        skipped_count += load_result.skipped
        if not load_result.success:
            errors.extend(load_result.errors)
            continue

        for fields in load_result.rows:
            lookup_value = self.get_lookup_value(fields)
            if lookup_value in (None, ""):
                skipped_count += 1
                continue

            _, created = model.objects.update_or_create(
                **{self.lookup_field: lookup_value},
                defaults=self.get_defaults(fields),
            )
            if created:
                created_count += 1
            else:
                updated_count += 1

    result = FixtureImportResult(
        created=created_count,
        updated=updated_count,
        skipped=skipped_count,
        errors=errors,
    )
    self.import_result = result

    if not result.success:
        for error in result.errors:
            self.stdout.write(self.style.ERROR(error))
        return False

    self.stdout.write(
        self.style.SUCCESS(
            f"Imported {self.fixture_key}: {result.created} created, "
            f"{result.updated} updated, {result.skipped} skipped"
        )
    )
    return True

SingletonFixtureUpdateCommand

Bases: BaseHashProtectedCommand

Update a singleton model instance from the first row in a JSON fixture.

Source code in src/codex_django/system/management/base_commands.py 
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
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
423
424
425
class SingletonFixtureUpdateCommand(BaseHashProtectedCommand):
    """Update a singleton model instance from the first row in a JSON fixture."""

    fixture_path: str | Path | None = None
    fixture_paths: ClassVar[list[str | Path]] = []
    model_path: str = ""
    model_class: Any = None
    singleton_pk: Any = 1
    fields_key: str = "fields"

    def get_fixture_paths(self) -> list[Path]:
        """Return configured fixture paths as :class:`Path` objects."""
        configured = list(self.fixture_paths)
        if self.fixture_path is not None:
            configured.append(self.fixture_path)
        return [Path(path) for path in configured]

    def get_model_class(self) -> Any:
        """Return the singleton model class."""
        if self.model_class is not None:
            return self.model_class
        if not self.model_path:
            raise ValueError("model_path or model_class is required")
        return apps.get_model(self.model_path)

    def get_singleton(self, model: Any) -> Any:
        """Return the singleton instance for the target model."""
        load = getattr(model, "load", None)
        if callable(load):
            return load()
        instance, _ = model.objects.get_or_create(pk=self.singleton_pk)
        return instance

    def normalize_current_value(self, field_name: str, value: Any) -> Any:
        """Normalize current model values before fixture comparison."""
        if hasattr(value, "__str__") and field_name in ["latitude", "longitude"]:
            return str(value)
        return value

    def get_update_fields(self, fixture_fields: dict[str, Any]) -> dict[str, Any]:
        """Return fixture fields eligible for singleton assignment."""
        return fixture_fields

    def sync_instance(self, instance: Any) -> None:
        """Synchronize the updated instance to Redis when a manager is available."""
        from codex_django.core.redis.managers.settings import get_site_settings_manager

        manager = get_site_settings_manager()
        manager.save_instance(instance)

    def handle_import(self, *args: Any, **options: Any) -> bool:
        """Apply the first fixture row to the configured singleton instance."""
        fixture_paths = self.get_fixture_paths()
        if not fixture_paths:
            self.stdout.write(self.style.ERROR(f"No fixture path configured for {self.fixture_key}."))
            return False

        load_result = load_json_fixture_rows(fixture_paths[0], fields_key=self.fields_key)
        if not load_result.success:
            for error in load_result.errors:
                self.stdout.write(self.style.ERROR(error))
            return False
        if not load_result.rows:
            self.stdout.write(self.style.ERROR(f"Invalid fixture format for {self.fixture_key}: no rows found."))
            return False

        model = self.get_model_class()
        instance = self.get_singleton(model)
        updated_fields: list[str] = []

        for field_name, new_value in self.get_update_fields(load_result.rows[0]).items():
            if not hasattr(instance, field_name):
                continue
            current_value = self.normalize_current_value(field_name, getattr(instance, field_name))
            if str(current_value) != str(new_value):
                setattr(instance, field_name, new_value)
                updated_fields.append(field_name)

        self.updated_fields = updated_fields
        if not updated_fields:
            self.stdout.write(self.style.SUCCESS(f"No changes needed for {self.fixture_key}."))
            return True

        instance.save()
        self.sync_instance(instance)
        self.stdout.write(self.style.SUCCESS(f"Updated {self.fixture_key}: {len(updated_fields)} fields changed"))
        return True
Functions
get_fixture_paths()

Return configured fixture paths as :class:Path objects.

Source code in src/codex_django/system/management/base_commands.py 
349
350
351
352
353
354
def get_fixture_paths(self) -> list[Path]:
    """Return configured fixture paths as :class:`Path` objects."""
    configured = list(self.fixture_paths)
    if self.fixture_path is not None:
        configured.append(self.fixture_path)
    return [Path(path) for path in configured]
get_model_class()

Return the singleton model class.

Source code in src/codex_django/system/management/base_commands.py 
356
357
358
359
360
361
362
def get_model_class(self) -> Any:
    """Return the singleton model class."""
    if self.model_class is not None:
        return self.model_class
    if not self.model_path:
        raise ValueError("model_path or model_class is required")
    return apps.get_model(self.model_path)
get_singleton(model)

Return the singleton instance for the target model.

Source code in src/codex_django/system/management/base_commands.py 
364
365
366
367
368
369
370
def get_singleton(self, model: Any) -> Any:
    """Return the singleton instance for the target model."""
    load = getattr(model, "load", None)
    if callable(load):
        return load()
    instance, _ = model.objects.get_or_create(pk=self.singleton_pk)
    return instance
normalize_current_value(field_name, value)

Normalize current model values before fixture comparison.

Source code in src/codex_django/system/management/base_commands.py 
372
373
374
375
376
def normalize_current_value(self, field_name: str, value: Any) -> Any:
    """Normalize current model values before fixture comparison."""
    if hasattr(value, "__str__") and field_name in ["latitude", "longitude"]:
        return str(value)
    return value
get_update_fields(fixture_fields)

Return fixture fields eligible for singleton assignment.

Source code in src/codex_django/system/management/base_commands.py 
378
379
380
def get_update_fields(self, fixture_fields: dict[str, Any]) -> dict[str, Any]:
    """Return fixture fields eligible for singleton assignment."""
    return fixture_fields
sync_instance(instance)

Synchronize the updated instance to Redis when a manager is available.

Source code in src/codex_django/system/management/base_commands.py 
382
383
384
385
386
387
def sync_instance(self, instance: Any) -> None:
    """Synchronize the updated instance to Redis when a manager is available."""
    from codex_django.core.redis.managers.settings import get_site_settings_manager

    manager = get_site_settings_manager()
    manager.save_instance(instance)
handle_import(*args, **options)

Apply the first fixture row to the configured singleton instance.

Source code in src/codex_django/system/management/base_commands.py 
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
423
424
425
def handle_import(self, *args: Any, **options: Any) -> bool:
    """Apply the first fixture row to the configured singleton instance."""
    fixture_paths = self.get_fixture_paths()
    if not fixture_paths:
        self.stdout.write(self.style.ERROR(f"No fixture path configured for {self.fixture_key}."))
        return False

    load_result = load_json_fixture_rows(fixture_paths[0], fields_key=self.fields_key)
    if not load_result.success:
        for error in load_result.errors:
            self.stdout.write(self.style.ERROR(error))
        return False
    if not load_result.rows:
        self.stdout.write(self.style.ERROR(f"Invalid fixture format for {self.fixture_key}: no rows found."))
        return False

    model = self.get_model_class()
    instance = self.get_singleton(model)
    updated_fields: list[str] = []

    for field_name, new_value in self.get_update_fields(load_result.rows[0]).items():
        if not hasattr(instance, field_name):
            continue
        current_value = self.normalize_current_value(field_name, getattr(instance, field_name))
        if str(current_value) != str(new_value):
            setattr(instance, field_name, new_value)
            updated_fields.append(field_name)

    self.updated_fields = updated_fields
    if not updated_fields:
        self.stdout.write(self.style.SUCCESS(f"No changes needed for {self.fixture_key}."))
        return True

    instance.save()
    self.sync_instance(instance)
    self.stdout.write(self.style.SUCCESS(f"Updated {self.fixture_key}: {len(updated_fields)} fields changed"))
    return True

Functions

load_json_fixture_rows(path, fields_key='fields')

Load a Django-style JSON fixture and return each item's fields mapping.

Parameters:

Name Type Description Default
path Path

Fixture file path.

 required
fields_key str

Item key containing model field values.

 'fields'

Returns:

Type Description
JsonFixtureLoadResult

Parsed row mappings plus skipped-item and error diagnostics. Invalid
JsonFixtureLoadResult

JSON, unreadable files, and non-list payloads are fatal errors. Items
JsonFixtureLoadResult

without a usable fields mapping are skipped.
Source code in src/codex_django/system/management/base_commands.py 
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
def load_json_fixture_rows(path: Path, fields_key: str = "fields") -> JsonFixtureLoadResult:
    """Load a Django-style JSON fixture and return each item's ``fields`` mapping.

    Args:
        path: Fixture file path.
        fields_key: Item key containing model field values.

    Returns:
        Parsed row mappings plus skipped-item and error diagnostics. Invalid
        JSON, unreadable files, and non-list payloads are fatal errors. Items
        without a usable fields mapping are skipped.
    """
    try:
        with open(path, encoding="utf-8") as f:
            payload = json.load(f)
    except json.JSONDecodeError as exc:
        return JsonFixtureLoadResult(errors=[f"Error decoding JSON fixture {path}: {exc}"])
    except OSError as exc:
        return JsonFixtureLoadResult(errors=[f"Error reading JSON fixture {path}: {exc}"])

    if not isinstance(payload, list):
        return JsonFixtureLoadResult(errors=[f"Invalid fixture format in {path}: expected a list."])

    rows: list[dict[str, Any]] = []
    skipped = 0
    for item in payload:
        if not isinstance(item, Mapping):
            skipped += 1
            continue
        fields_value = item.get(fields_key)
        if not isinstance(fields_value, Mapping):
            skipped += 1
            continue
        rows.append(dict(fields_value))

    return JsonFixtureLoadResult(rows=rows, skipped=skipped)

Fixture Redis manager

codex_django.system.redis.managers.fixtures

Redis-backed helpers for fixture hash tracking.

Classes

FixtureHashManager

Bases: BaseDjangoRedisManager

Store fixture content hashes to skip redundant import operations.

Notes

Each logical fixture group is stored under a fixture:<key> Redis string so import commands can quickly decide whether work is needed.

Source code in src/codex_django/system/redis/managers/fixtures.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
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
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
class FixtureHashManager(BaseDjangoRedisManager):
    """Store fixture content hashes to skip redundant import operations.

    Notes:
        Each logical fixture group is stored under a ``fixture:<key>`` Redis
        string so import commands can quickly decide whether work is needed.
    """

    async def aget_hash(self, fixture_key: str) -> str | None:
        """Return the stored hash for a logical fixture group, if any.

        Args:
            fixture_key: Logical identifier for the fixture bundle.

        Returns:
            The stored hash string or ``None`` when no value exists.
        """
        if self._is_disabled():
            return None
        async with self.async_string() as s:
            return await s.get(self.make_key(f"fixture:{fixture_key}"))  # type: ignore[no-any-return]

    async def aset_hash(self, fixture_key: str, hash_string: str) -> None:
        """Persist the latest hash for a logical fixture group.

        Args:
            fixture_key: Logical identifier for the fixture bundle.
            hash_string: Newly computed content hash to store.
        """
        if self._is_disabled():
            return
        async with self.async_string() as s:
            await s.set(self.make_key(f"fixture:{fixture_key}"), hash_string)

    def get_hash(self, fixture_key: str) -> str | None:
        """Synchronously return the stored hash for a fixture bundle.

        Args:
            fixture_key: Logical identifier for the fixture bundle.

        Returns:
            The stored hash string or ``None`` when no value exists.
        """
        if self._is_disabled():
            return None
        with self.sync_string() as s:
            return s.get(self.make_key(f"fixture:{fixture_key}"))  # type: ignore[no-any-return]

    def set_hash(self, fixture_key: str, hash_string: str) -> None:
        """Synchronously persist a fixture hash value.

        Args:
            fixture_key: Logical identifier for the fixture bundle.
            hash_string: Newly computed content hash to store.
        """
        if self._is_disabled():
            return
        with self.sync_string() as s:
            s.set(self.make_key(f"fixture:{fixture_key}"), hash_string)
Functions
aget_hash(fixture_key) async

Return the stored hash for a logical fixture group, if any.

Parameters:

Name Type Description Default
fixture_key str

Logical identifier for the fixture bundle.

 required

Returns:

Type Description
str | None

The stored hash string or None when no value exists.
Source code in src/codex_django/system/redis/managers/fixtures.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
async def aget_hash(self, fixture_key: str) -> str | None:
    """Return the stored hash for a logical fixture group, if any.

    Args:
        fixture_key: Logical identifier for the fixture bundle.

    Returns:
        The stored hash string or ``None`` when no value exists.
    """
    if self._is_disabled():
        return None
    async with self.async_string() as s:
        return await s.get(self.make_key(f"fixture:{fixture_key}"))  # type: ignore[no-any-return]
aset_hash(fixture_key, hash_string) async

Persist the latest hash for a logical fixture group.

Parameters:

Name Type Description Default
fixture_key str

Logical identifier for the fixture bundle.

 required
hash_string str

Newly computed content hash to store.

 required
Source code in src/codex_django/system/redis/managers/fixtures.py 
28
29
30
31
32
33
34
35
36
37
38
async def aset_hash(self, fixture_key: str, hash_string: str) -> None:
    """Persist the latest hash for a logical fixture group.

    Args:
        fixture_key: Logical identifier for the fixture bundle.
        hash_string: Newly computed content hash to store.
    """
    if self._is_disabled():
        return
    async with self.async_string() as s:
        await s.set(self.make_key(f"fixture:{fixture_key}"), hash_string)
get_hash(fixture_key)

Synchronously return the stored hash for a fixture bundle.

Parameters:

Name Type Description Default
fixture_key str

Logical identifier for the fixture bundle.

 required

Returns:

Type Description
str | None

The stored hash string or None when no value exists.
Source code in src/codex_django/system/redis/managers/fixtures.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
def get_hash(self, fixture_key: str) -> str | None:
    """Synchronously return the stored hash for a fixture bundle.

    Args:
        fixture_key: Logical identifier for the fixture bundle.

    Returns:
        The stored hash string or ``None`` when no value exists.
    """
    if self._is_disabled():
        return None
    with self.sync_string() as s:
        return s.get(self.make_key(f"fixture:{fixture_key}"))  # type: ignore[no-any-return]
set_hash(fixture_key, hash_string)

Synchronously persist a fixture hash value.

Parameters:

Name Type Description Default
fixture_key str

Logical identifier for the fixture bundle.

 required
hash_string str

Newly computed content hash to store.

 required
Source code in src/codex_django/system/redis/managers/fixtures.py 
54
55
56
57
58
59
60
61
62
63
64
def set_hash(self, fixture_key: str, hash_string: str) -> None:
    """Synchronously persist a fixture hash value.

    Args:
        fixture_key: Logical identifier for the fixture bundle.
        hash_string: Newly computed content hash to store.
    """
    if self._is_disabled():
        return
    with self.sync_string() as s:
        s.set(self.make_key(f"fixture:{fixture_key}"), hash_string)

Functions

get_fixture_hash_manager()

Return a fixture hash manager configured from Django settings.

Returns:

Type Description
FixtureHashManager

A ready-to-use :class:FixtureHashManager instance.
Source code in src/codex_django/system/redis/managers/fixtures.py 
67
68
69
70
71
72
73
def get_fixture_hash_manager() -> FixtureHashManager:
    """Return a fixture hash manager configured from Django settings.

    Returns:
        A ready-to-use :class:`FixtureHashManager` instance.
    """
    return FixtureHashManager()

Action token Redis manager

codex_django.system.redis.managers.tokens

Redis-backed JSON action token helpers.

Classes

JsonActionTokenRedisManager

Bases: BaseDjangoRedisManager

Store temporary JSON payloads behind secure action tokens.

Projects can subclass this manager to type or validate their own payload shapes while reusing token generation, Redis keying, TTL, decode, and cleanup behavior.

Source code in src/codex_django/system/redis/managers/tokens.py 
 13
 14
 15
 16
 17
 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
 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
185
186
187
188
class JsonActionTokenRedisManager(BaseDjangoRedisManager):
    """Store temporary JSON payloads behind secure action tokens.

    Projects can subclass this manager to type or validate their own payload
    shapes while reusing token generation, Redis keying, TTL, decode, and
    cleanup behavior.
    """

    default_ttl_seconds = 24 * 60 * 60
    token_bytes = 32

    def __init__(self, prefix: str = "auth:action", **kwargs: Any) -> None:
        """Initialize the token manager.

        Args:
            prefix: Redis key prefix used for stored token payloads.
            **kwargs: Additional keyword arguments forwarded to
                ``BaseDjangoRedisManager``.
        """
        super().__init__(prefix=prefix, **kwargs)

    def make_token(self) -> str:
        """Return a random URL-safe token string."""
        return secrets.token_urlsafe(self.token_bytes)

    def validate_payload(self, payload: Mapping[str, Any]) -> dict[str, Any]:
        """Return a JSON-serializable payload mapping.

        Subclasses may override this to enforce project-specific schemas.
        """
        return dict(payload)

    async def acreate_token(
        self,
        payload: Mapping[str, Any],
        *,
        ttl_seconds: int | None = None,
        ttl_hours: int | None = None,
    ) -> str:
        """Create a token and store its payload with a TTL.

        Args:
            payload: Mapping to serialize and store behind the generated token.
            ttl_seconds: Optional TTL override in seconds.
            ttl_hours: Optional TTL override in hours. Used only when
                ``ttl_seconds`` is not provided.

        Returns:
            The generated URL-safe action token.
        """
        token = self.make_token()
        if self._is_disabled():
            return token

        timeout = ttl_seconds
        if timeout is None and ttl_hours is not None:
            timeout = ttl_hours * 60 * 60
        if timeout is None:
            timeout = self.default_ttl_seconds

        async with self.async_string() as s:
            await s.set(
                self.make_key(token), json.dumps(self.validate_payload(payload), cls=DjangoJSONEncoder), ttl=timeout
            )
        return token

    async def aget_token_data(self, token: str) -> dict[str, Any] | None:
        """Return decoded token payload data when present and valid.

        Args:
            token: Action token previously returned by ``acreate_token()`` or
                ``create_token()``.

        Returns:
            The decoded JSON payload mapping, or ``None`` when the token is
            missing, expired, disabled, or contains invalid JSON data.
        """
        if self._is_disabled():
            return None

        async with self.async_string() as s:
            raw_data = await s.get(self.make_key(token))

        if not raw_data:
            return None
        try:
            data = json.loads(raw_data)
        except (json.JSONDecodeError, TypeError):
            return None
        if not isinstance(data, dict):
            return None
        return data

    async def adelete_token(self, token: str) -> None:
        """Delete a token payload.

        Args:
            token: Action token whose stored payload should be removed.
        """
        if self._is_disabled():
            return
        async with self.async_string() as s:
            await s.delete(self.make_key(token))

    def create_token(
        self,
        payload: Mapping[str, Any],
        *,
        ttl_seconds: int | None = None,
        ttl_hours: int | None = None,
    ) -> str:
        """Synchronously create a token and store its payload.

        Args:
            payload: Mapping to serialize and store behind the generated token.
            ttl_seconds: Optional TTL override in seconds.
            ttl_hours: Optional TTL override in hours. Used only when
                ``ttl_seconds`` is not provided.

        Returns:
            The generated URL-safe action token.
        """
        token = self.make_token()
        if self._is_disabled():
            return token

        timeout = ttl_seconds
        if timeout is None and ttl_hours is not None:
            timeout = ttl_hours * 60 * 60
        if timeout is None:
            timeout = self.default_ttl_seconds

        with self.sync_string() as s:
            s.set(
                self.make_key(token),
                json.dumps(self.validate_payload(payload), cls=DjangoJSONEncoder),
                ttl=timeout,
            )
        return token

    def get_token_data(self, token: str) -> dict[str, Any] | None:
        """Synchronously return decoded token payload data.

        Args:
            token: Action token previously returned by ``create_token()``.

        Returns:
            The decoded payload mapping, or ``None`` when the token is
            missing, expired, disabled, or invalid.
        """
        if self._is_disabled():
            return None

        with self.sync_string() as s:
            raw_data = s.get(self.make_key(token))

        if not raw_data:
            return None
        try:
            data = json.loads(raw_data)
        except (json.JSONDecodeError, TypeError):
            return None
        if not isinstance(data, dict):
            return None
        return data

    def delete_token(self, token: str) -> None:
        """Synchronously delete token payload data.

        Args:
            token: Action token whose stored payload should be removed.
        """
        if self._is_disabled():
            return
        with self.sync_string() as s:
            s.delete(self.make_key(token))
Functions
__init__(prefix='auth:action', **kwargs)

Initialize the token manager.

Parameters:

Name Type Description Default
prefix str

Redis key prefix used for stored token payloads.

 'auth:action'
**kwargs Any

Additional keyword arguments forwarded to BaseDjangoRedisManager.

 {}
Source code in src/codex_django/system/redis/managers/tokens.py 
24
25
26
27
28
29
30
31
32
def __init__(self, prefix: str = "auth:action", **kwargs: Any) -> None:
    """Initialize the token manager.

    Args:
        prefix: Redis key prefix used for stored token payloads.
        **kwargs: Additional keyword arguments forwarded to
            ``BaseDjangoRedisManager``.
    """
    super().__init__(prefix=prefix, **kwargs)
make_token()

Return a random URL-safe token string.

Source code in src/codex_django/system/redis/managers/tokens.py 
34
35
36
def make_token(self) -> str:
    """Return a random URL-safe token string."""
    return secrets.token_urlsafe(self.token_bytes)
validate_payload(payload)

Return a JSON-serializable payload mapping.

Subclasses may override this to enforce project-specific schemas.

Source code in src/codex_django/system/redis/managers/tokens.py 
38
39
40
41
42
43
def validate_payload(self, payload: Mapping[str, Any]) -> dict[str, Any]:
    """Return a JSON-serializable payload mapping.

    Subclasses may override this to enforce project-specific schemas.
    """
    return dict(payload)
acreate_token(payload, *, ttl_seconds=None, ttl_hours=None) async

Create a token and store its payload with a TTL.

Parameters:

Name Type Description Default
payload Mapping[str, Any]

Mapping to serialize and store behind the generated token.

 required
ttl_seconds int | None

Optional TTL override in seconds.

 None
ttl_hours int | None

Optional TTL override in hours. Used only when ttl_seconds is not provided.

 None

Returns:

Type Description
str

The generated URL-safe action token.
Source code in src/codex_django/system/redis/managers/tokens.py 
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
async def acreate_token(
    self,
    payload: Mapping[str, Any],
    *,
    ttl_seconds: int | None = None,
    ttl_hours: int | None = None,
) -> str:
    """Create a token and store its payload with a TTL.

    Args:
        payload: Mapping to serialize and store behind the generated token.
        ttl_seconds: Optional TTL override in seconds.
        ttl_hours: Optional TTL override in hours. Used only when
            ``ttl_seconds`` is not provided.

    Returns:
        The generated URL-safe action token.
    """
    token = self.make_token()
    if self._is_disabled():
        return token

    timeout = ttl_seconds
    if timeout is None and ttl_hours is not None:
        timeout = ttl_hours * 60 * 60
    if timeout is None:
        timeout = self.default_ttl_seconds

    async with self.async_string() as s:
        await s.set(
            self.make_key(token), json.dumps(self.validate_payload(payload), cls=DjangoJSONEncoder), ttl=timeout
        )
    return token
aget_token_data(token) async

Return decoded token payload data when present and valid.

Parameters:

Name Type Description Default
token str

Action token previously returned by acreate_token() or create_token().

 required

Returns:

Type Description
dict[str, Any] | None

The decoded JSON payload mapping, or None when the token is
dict[str, Any] | None

missing, expired, disabled, or contains invalid JSON data.
Source code in src/codex_django/system/redis/managers/tokens.py 
 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
async def aget_token_data(self, token: str) -> dict[str, Any] | None:
    """Return decoded token payload data when present and valid.

    Args:
        token: Action token previously returned by ``acreate_token()`` or
            ``create_token()``.

    Returns:
        The decoded JSON payload mapping, or ``None`` when the token is
        missing, expired, disabled, or contains invalid JSON data.
    """
    if self._is_disabled():
        return None

    async with self.async_string() as s:
        raw_data = await s.get(self.make_key(token))

    if not raw_data:
        return None
    try:
        data = json.loads(raw_data)
    except (json.JSONDecodeError, TypeError):
        return None
    if not isinstance(data, dict):
        return None
    return data
adelete_token(token) async

Delete a token payload.

Parameters:

Name Type Description Default
token str

Action token whose stored payload should be removed.

 required
Source code in src/codex_django/system/redis/managers/tokens.py 
106
107
108
109
110
111
112
113
114
115
async def adelete_token(self, token: str) -> None:
    """Delete a token payload.

    Args:
        token: Action token whose stored payload should be removed.
    """
    if self._is_disabled():
        return
    async with self.async_string() as s:
        await s.delete(self.make_key(token))
create_token(payload, *, ttl_seconds=None, ttl_hours=None)

Synchronously create a token and store its payload.

Parameters:

Name Type Description Default
payload Mapping[str, Any]

Mapping to serialize and store behind the generated token.

 required
ttl_seconds int | None

Optional TTL override in seconds.

 None
ttl_hours int | None

Optional TTL override in hours. Used only when ttl_seconds is not provided.

 None

Returns:

Type Description
str

The generated URL-safe action token.
Source code in src/codex_django/system/redis/managers/tokens.py 
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
def create_token(
    self,
    payload: Mapping[str, Any],
    *,
    ttl_seconds: int | None = None,
    ttl_hours: int | None = None,
) -> str:
    """Synchronously create a token and store its payload.

    Args:
        payload: Mapping to serialize and store behind the generated token.
        ttl_seconds: Optional TTL override in seconds.
        ttl_hours: Optional TTL override in hours. Used only when
            ``ttl_seconds`` is not provided.

    Returns:
        The generated URL-safe action token.
    """
    token = self.make_token()
    if self._is_disabled():
        return token

    timeout = ttl_seconds
    if timeout is None and ttl_hours is not None:
        timeout = ttl_hours * 60 * 60
    if timeout is None:
        timeout = self.default_ttl_seconds

    with self.sync_string() as s:
        s.set(
            self.make_key(token),
            json.dumps(self.validate_payload(payload), cls=DjangoJSONEncoder),
            ttl=timeout,
        )
    return token
get_token_data(token)

Synchronously return decoded token payload data.

Parameters:

Name Type Description Default
token str

Action token previously returned by create_token().

 required

Returns:

Type Description
dict[str, Any] | None

The decoded payload mapping, or None when the token is
dict[str, Any] | None

missing, expired, disabled, or invalid.
Source code in src/codex_django/system/redis/managers/tokens.py 
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
def get_token_data(self, token: str) -> dict[str, Any] | None:
    """Synchronously return decoded token payload data.

    Args:
        token: Action token previously returned by ``create_token()``.

    Returns:
        The decoded payload mapping, or ``None`` when the token is
        missing, expired, disabled, or invalid.
    """
    if self._is_disabled():
        return None

    with self.sync_string() as s:
        raw_data = s.get(self.make_key(token))

    if not raw_data:
        return None
    try:
        data = json.loads(raw_data)
    except (json.JSONDecodeError, TypeError):
        return None
    if not isinstance(data, dict):
        return None
    return data
delete_token(token)

Synchronously delete token payload data.

Parameters:

Name Type Description Default
token str

Action token whose stored payload should be removed.

 required
Source code in src/codex_django/system/redis/managers/tokens.py 
179
180
181
182
183
184
185
186
187
188
def delete_token(self, token: str) -> None:
    """Synchronously delete token payload data.

    Args:
        token: Action token whose stored payload should be removed.
    """
    if self._is_disabled():
        return
    with self.sync_string() as s:
        s.delete(self.make_key(token))

Functions

get_json_action_token_manager(prefix='auth:action')

Return a JSON action token manager.

Parameters:

Name Type Description Default
prefix str

Redis key prefix used for stored token payloads.

 'auth:action'

Returns:

Type Description
JsonActionTokenRedisManager

A JsonActionTokenRedisManager configured with the requested key
JsonActionTokenRedisManager

prefix.
Source code in src/codex_django/system/redis/managers/tokens.py 
191
192
193
194
195
196
197
198
199
200
201
def get_json_action_token_manager(prefix: str = "auth:action") -> JsonActionTokenRedisManager:
    """Return a JSON action token manager.

    Args:
        prefix: Redis key prefix used for stored token payloads.

    Returns:
        A ``JsonActionTokenRedisManager`` configured with the requested key
        prefix.
    """
    return JsonActionTokenRedisManager(prefix=prefix)

Fixture hashing helpers

codex_django.system.utils.fixture_hash

Hash helpers for content and fixture synchronization workflows.

Functions

compute_file_hash(path, chunk_size=8192)

Compute a SHA-256 digest for a single file.

Parameters:

Name Type Description Default
path Path

Filesystem path to the file that should be hashed.

 required
chunk_size int

Read size used while streaming file contents.

 8192

Returns:

Type Description
str

Hex-encoded SHA-256 digest for the file contents.
Source code in src/codex_django/system/utils/fixture_hash.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
def compute_file_hash(path: Path, chunk_size: int = 8192) -> str:
    """Compute a SHA-256 digest for a single file.

    Args:
        path: Filesystem path to the file that should be hashed.
        chunk_size: Read size used while streaming file contents.

    Returns:
        Hex-encoded SHA-256 digest for the file contents.
    """
    sha256 = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(chunk_size), b""):
            sha256.update(chunk)
    return sha256.hexdigest()

compute_paths_hash(paths)

Compute a combined SHA-256 digest for multiple fixture files.

The digest incorporates both file names and file contents so that renames and content changes invalidate the stored value. Non-file paths are ignored.

Parameters:

Name Type Description Default
paths list[Path]

Candidate filesystem paths to include in the combined hash.

 required

Returns:

Type Description
str

Hex-encoded SHA-256 digest for the ordered set of existing files.
Source code in src/codex_django/system/utils/fixture_hash.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def compute_paths_hash(paths: list[Path]) -> str:
    """Compute a combined SHA-256 digest for multiple fixture files.

    The digest incorporates both file names and file contents so that renames
    and content changes invalidate the stored value. Non-file paths are
    ignored.

    Args:
        paths: Candidate filesystem paths to include in the combined hash.

    Returns:
        Hex-encoded SHA-256 digest for the ordered set of existing files.
    """
    sha256 = hashlib.sha256()
    for p in sorted(paths, key=lambda x: x.name):
        if not p.is_file():
            continue
        sha256.update(p.name.encode("utf-8"))
        with open(p, "rb") as f:
            for chunk in iter(lambda: f.read(8192), b""):
                sha256.update(chunk)
    return sha256.hexdigest()