ArchiveBox 基础模型层解析基于 UUIDv7 主键与可复用 Django Mixin 的设计实践【免费下载链接】ArchiveBox Open source self-hosted web archiving. Takes URLs/browser history/bookmarks/Pocket/Pinboard/etc., saves HTML, JS, PDFs, media, and more...项目地址: https://gitcode.com/gh_mirrors/ar/ArchiveBox本文围绕archivebox/base_models/models.py展开这是 ArchiveBox 全项目 Django ORM 模型的公共地基。它定义了ModelWithUUID、ModelWithNotes、ModelWithHealthStats、ModelWithConfig、ModelWithDeleteAfter、ModelWithOutputDir六个抽象 Mixin 与AutoDateTimeField字段外加两个模块级工具函数为 Snapshot、Crawl、ArchiveResult、Process 等核心模型统一提供 UUIDv7 主键、审计时间戳、健康统计、JSON 配置、自动清理与磁盘输出目录管理能力。读完本文你将掌握该模块的每个类与函数的实现细节、它们如何被组合进具体业务模型以及如何在自己的 Django 项目中复刻这套模型基建。模块定位全项目模型的公共地基archivebox/base_models/models.py的文件头注释一句话点明其使命Base models using UUIDv7 for all id fields所有 id 字段使用 UUIDv7 的基础模型。该模块位于 archivebox/base_models/models.py被多个 Django app 引用core应用Snapshot、ArchiveResult、Tag等核心归档模型crawls应用Crawl、CrawlSchedulemachine应用Machine、NetworkInterface、Binary、Processpersonas应用Persona。从模块目录看它对外暴露的公共 API 由两部分组成函数2 个normalize_config_json_values(config) - Anyget_or_create_system_user_pk(usernamesystem)类7 个AutoDateTimeField继承django.db.models.DateTimeFieldModelWithUUID继承django.db.models.ModelModelWithNotesModelWithHealthStatsModelWithConfigModelWithDeleteAfterModelWithOutputDir继承ModelWithUUID除AutoDateTimeField外其余六个类均为Meta.abstract True的抽象模型作为 Mixin 被业务模型多重继承组合使用。这种一个字段能力一个 Mixin的拆分方式使得不同业务模型可以按需拼装能力避免重复代码。ModelWithUUIDUUIDv7 主键模型基类ModelWithUUID是模块中最核心的基类为所有继承它的模型提供统一的身份与审计字段。其定义在 archivebox/base_models/models.pyclass ModelWithUUID(models.Model): id CompactUUIDField(primary_keyTrue, defaultuuid7, editableFalse, uniqueTrue) created_at models.DateTimeField(defaulttimezone.now, db_indexTrue) modified_at models.DateTimeField(auto_nowTrue) created_by models.ForeignKey( settings.AUTH_USER_MODEL, on_deletemodels.CASCADE, defaultget_or_create_system_user_pk, nullFalse, db_indexTrue, ) class Meta(TypedModelMeta): abstract True def __str__(self) - str: return f[{self.id}] {self.__class__.__name__}UUIDv7 主键时间有序、无中心协调id字段使用CompactUUIDField(primary_keyTrue, defaultuuid7, editableFalse, uniqueTrue)。其中uuid7是模块内定义的主键默认值生成器见下文UUID7 兼容层CompactUUIDField继承自 Django 的models.UUIDField重写了to_python、from_db_value与deconstruct见 archivebox/uuid_compat.py将 UUID 以无连字符的紧凑十六进制字符串形式存储与展示。选择 UUIDv7 而不是数据库自增主键或随机 UUIDv4关键在于UUIDv7 前缀包含时间戳天然按创建时间排序。这带来两个直接收益批量插入友好UUIDv7 大致随时间单调递增比 UUIDv4 的随机主键更有利于索引局部性与 B 树页填充无需中央协调可以安全地在采集器、后台进程、CLI 等多进程环境中离线生成主键不会发生自增主键常见的竞态。配合created_at默认timezone.now带db_index与modified_atauto_nowTrue每次 save 自动更新每个模型天然拥有完整的时间审计信息。created_by系统操作用户的兜底方案created_by外键指向settings.AUTH_USER_MODELon_deletemodels.CASCADE其默认值由模块级函数get_or_create_system_user_pk提供。该函数在 archivebox/base_models/models.pydef get_or_create_system_user_pk(usernamesystem): User get_user_model() # If theres exactly one superuser, use that for all system operations if User.objects.filter(is_superuserTrue).count() 1: return User.objects.filter(is_superuserTrue).values_list(pk, flatTrue)[0] # Otherwise get or create the system user user, _ User.objects.get_or_create( usernameusername, defaults{is_staff: True, is_superuser: True, email: , password: !}, ) return user.pk其设计意图非常明确唯一超级用户优先若系统中恰好只有一个 superuser多数单机自托管部署的典型形态所有系统自动创建的行如爬虫调度的 Crawl、后台进程写入的 Process都归到该用户名下避免额外创建系统账号否则兜底创建system用户当超级用户数量不为 1多个或零个时get_or_create一个用户名为system的账号is_staffTrue、is_superuserTrue、密码为无效的!并返回其主键。这一设计解决了自动任务产生的行记录由谁创建的问题created_by永远不会为 NULLnullFalse同时又能避免批量创建无意义的用户记录。从源码结构看personas/models.py与crawls/models.py也直接引用了该函数作为外键默认值。便捷 URL 属性ModelWithUUID还提供三个只读属性方便在后台管理与 API 场景快速构造链接见 archivebox/base_models/models.pyproperty def admin_change_url(self) - str: return f/admin/{self._meta.app_label}/{self._meta.model_name}/{self.pk}/change/ property def api_url(self) - str: return str(reverse_lazy(api-1:get_any, args[self.id])) property def api_docs_url(self) - str: return f/api/v1/docs#/{self._meta.app_label.title()}%20Models/api_v1_{self._meta.app_label}_get_{self._meta.db_table}admin_change_url通过 Django admin 的app_label、model_name与主键拼出后台编辑页地址api_url利用命名路由api-1:get_any反解出 v1 API 的通用对象端点ArchiveBox 的 v1 API 支持按任意模型 ID 查询资源参见 archivebox/api/urls.pyapi_docs_url指向 Swagger/OpenAPI 文档中该模型的条目锚点。这三个属性让模板与序列化器无需关心路由细节即可输出标准链接。AutoDateTimeField旧版自动时间戳兼容字段AutoDateTimeField继承django.db.models.DateTimeField其 docstring 明确说明用途DateTimeField that automatically updates on save (legacy compatibility)即为旧版数据提供兼容性的自动更新时间字段。实现位于 archivebox/base_models/models.pyclass AutoDateTimeField(models.DateTimeField): DateTimeField that automatically updates on save (legacy compatibility). def pre_save(self, model_instance, add): if add or self.attname not in model_instance.__dict__ or not model_instance.__dict__[self.attname]: value timezone.now() setattr(model_instance, self.attname, value) return value return super().pre_save(model_instance, add)pre_save是 Django 字段在Model.save()之前调用的钩子。该字段的行为逻辑是若正在新增记录addTrue、或实例中尚无该字段值、或该值为空则强制填充当前时间timezone.now()否则走父类默认逻辑保留已存在的值。与ModelWithUUID.modified_atauto_nowTrue的区别在于AutoDateTimeField允许在特定场景下保留旧值主要用于历史数据迁移与旧版数据模型的兼容读取。模块级函数normalize_config_json_valuesnormalize_config_json_values(config)是ModelWithConfig.save()的核心依赖负责清洗 JSON 配置中的字符串值。实现见 archivebox/base_models/models.pydef normalize_config_json_values(config: Any) - Any: if not isinstance(config, dict): return config normalized dict(config) for key, value in list(normalized.items()): if not isinstance(value, str) or len(value) 2: continue if value[:1] ! or value[-1:] ! : continue try: decoded json.loads(value) except ValueError: continue if isinstance(decoded, str): normalized[key] decoded return normalized它的工作方式非 dict 值直接原样返回遍历每个键值对只处理长度 ≥ 2、以双引号开头且以双引号结尾的字符串尝试用json.loads解析这段引号包裹的字符串若解析成功且结果为字符串则把该值替换为解包后的纯字符串。典型场景当配置值经过 JSON 序列化-反序列化往返后字符串可能被二次编码成\foo\这样的嵌套引号形式。normalize_config_json_values负责把这些字符串化的字符串还原为干净的值保证configJSON 字段中的数据始终规整、可预测。从源码结构看该函数也被 archivebox/machine/models.py 直接导入使用。ModelWithNotes备注字段 MixinModelWithNotes是最轻量的 Mixin只提供一个字段见 archivebox/base_models/models.pyclass ModelWithNotes(models.Model): Mixin for models with a notes field. notes models.TextField(blankTrue, nullFalse, default) class Meta(TypedModelMeta): abstract Truenotes是允许为空blankTrue但非 NULLnullFalse、默认的TextField用于记录用户备注或系统说明。它被Snapshot、ArchiveResult、Crawl、CrawlSchedule等模型继承例如 archivebox/crawls/models.py 中class CrawlSchedule(ModelWithUUID, ModelWithNotes)。ModelWithHealthStats健康统计与原子计数ModelWithHealthStats为模型提供失败次数 / 成功次数两个统计字段与一个健康度百分比计算属性见 archivebox/base_models/models.pyclass ModelWithHealthStats(models.Model): Mixin for models with health tracking fields. num_uses_failed models.PositiveIntegerField(default0) num_uses_succeeded models.PositiveIntegerField(default0) class Meta(TypedModelMeta): abstract True property def admin_change_url(self) - str: return f/admin/{self._meta.app_label}/{self._meta.model_name}/{self.pk}/change/ property def health(self) - int: total max(self.num_uses_failed self.num_uses_succeeded, 1) return round((self.num_uses_succeeded / total) * 100) def increment_health_stats(self, success: bool): Atomically increment success or failure counter using F() expression. field num_uses_succeeded if success else num_uses_failed type(self).objects.filter(pkself.pk).update( **{ field: F(field) 1, modified_at: timezone.now(), }, )值得注意的实现细节health属性返回0~100的整数百分比num_uses_succeeded / (失败 成功)分母通过max(..., 1)兜底避免除零increment_health_stats(success)使用F()表达式原子递增。F(field) 1会把递增操作下推为 SQL 层的SET num_uses_succeeded num_uses_succeeded 1避免读-改-写三步在并发下丢失更新同时顺手刷新modified_at。这是 Django 中做并发安全计数器的标准做法该 Mixin 同时重定义了admin_change_url因此任何同时继承ModelWithUUID与ModelWithHealthStats的模型如Machine、Binary都会得到一致的属性。在业务模型中的实际使用者包括 archivebox/machine/models.py 的Machine、NetworkInterface、Binary以及Snapshot、Crawl等。对于机器、二进制依赖等会被反复调用且可能失败的实体健康统计可以直观反映其可用性。ModelWithConfigJSON 配置字段与写入规范化ModelWithConfig为模型挂载一个 JSON 配置字段并在save()时自动做值规范化见 archivebox/base_models/models.pyclass ModelWithConfig(models.Model): Mixin for models with a JSON config field. config models.JSONField(defaultdict, nullTrue, blankTrue, editableTrue) class Meta(TypedModelMeta): abstract True def save(self, *args, **kwargs): normalized_config normalize_config_json_values(self.config) if normalized_config ! self.config: self.config normalized_config update_fields kwargs.get(update_fields) if update_fields is not None: kwargs[update_fields] tuple(dict.fromkeys([*update_fields, config])) super().save(*args, **kwargs)要点config为 Django 原生JSONField默认dict允许 NULL 与空值可在 admin 中直接编辑save()覆写写入前调用normalize_config_json_values清理嵌套引号字符串若发生了规范化且调用方通过update_fields指定了部分更新字段则用dict.fromkeys去重后把config追加进update_fields确保规范化结果被真正落库这一机制保证了配置在两次读写之间保持幂等与稳定避免存进去是value读出来是value的脏数据。实际使用者包括Snapshot、Crawlarchivebox/crawls/models.py以及Personaarchivebox/personas/models.py 中class Persona(ModelWithConfig)。其中 Snapshot/Crawl 的config承担着冻结本次爬取的配置快照职责ArchiveBox 通过迁移 0018freeze_crawl_config_snapshots将调度时的配置固化到每行保证历史快照可复现见 archivebox/crawls/migrations/0018_freeze_crawl_config_snapshots.py。ModelWithDeleteAfter基于 DELETE_AFTER 的自动过期删除ModelWithDeleteAfter是模块中最具业务复杂度的 Mixin它实现了 ArchiveBox 的保留策略retention policy配置DELETE_AFTER之后到期记录会被自动清理。完整实现见 archivebox/base_models/models.py。字段与类属性class ModelWithDeleteAfter(models.Model): delete_after_final_statuses: tuple[str, ...] () delete_at models.DateTimeField(defaultNone, nullTrue, blankTrue, db_indexTrue)delete_after_final_statuses类级属性声明哪些终态final status记录才允许被删除。默认空元组表示不限制子类覆盖此值来限定范围见下文delete_atDateTimeField可为 NULL带db_index是过期删除的判据。save 钩子与配置解析def save(self, *args, **kwargs): update_fields kwargs.get(update_fields) if self.delete_at is None: self.set_delete_at_from_config() if self.delete_at is not None and update_fields is not None: kwargs[update_fields] tuple(dict.fromkeys([*update_fields, delete_at])) super().save(*args, **kwargs) def get_delete_after_config_value(self): from archivebox.config.common import get_config return get_config(include_machineFalse, resolve_pluginsFalse).DELETE_AFTER def set_delete_at_from_config(self, config_valueNone) - bool: if self.delete_at is not None: return False from archivebox.config.common import parse_delete_after duration parse_delete_after(self.get_delete_after_config_value() if config_value is None else config_value) if duration is None: return False self.delete_at (self.created_at or timezone.now()) duration return True核心逻辑save()覆写当delete_at为空时自动调用set_delete_at_from_config()依据全局配置计算删除时间点同样会感知update_fields并把delete_at加入部分更新集合get_delete_after_config_value()读取全局配置中的DELETE_AFTER。基类默认直接取get_config(...).DELETE_AFTER子类如Snapshot、Crawl、Process会覆写它改为从自身config或父级如 Snapshot 的所属 Crawl的配置中解析实现逐行/逐级可覆盖的保留策略。例如 archivebox/core/models.py 中Snapshot.get_delete_after_config_value调用了resolve_delete_after_config_value(self.config, self.crawl.config)将 Snapshot 自身配置与所属 Crawl 配置合并解析set_delete_at_from_config()把时长解析为timedelta叠加在created_at无则用当前时间上得到delete_at返回是否成功设置。DELETE_AFTER 时长格式时长解析函数parse_delete_after定义在 archivebox/config/common.py它是整个保留策略的语法入口取值含义示例0//none/false/no/off禁用自动删除DELETE_AFTER0h/hr/hrs/hour/hours小时DELETE_AFTER2hd/day/days天DELETE_AFTER7dw/week/weeks周DELETE_AFTER4wmo/month/months月按 30 天计DELETE_AFTER6moy/yr/yrs/year/years年按 365 天计DELETE_AFTER1y规则细节语法为(\d)\s*(单位)的正则全匹配非法格式直接抛ValueError非零时长最短为 1 小时duration timedelta(hours1)会报错该配置在ArchiveConfig中的定义见 archivebox/config/common.py字段校验器validate_delete_after会在配置加载阶段提前做合法性校验。delete_expired批量过期清理classmethod def delete_expired(cls, *, batch_size: int 100, backfill_missing: bool True) - int: if backfill_missing: missing_delete_at list(cls.missing_delete_at_candidates().order_by(created_at, pk)[:batch_size]) for obj in missing_delete_at: if obj.set_delete_at_from_config(): cls.objects.filter(pkobj.pk, delete_at__isnullTrue).update( delete_atobj.delete_at, modified_attimezone.now(), ) # Keep the expiration sweep anchored on delete_at. Some large tables # have millions of final-status rows but almost no retained rows; ... due_pks list( cls.objects.filter(delete_at__isnullFalse, delete_at__ltetimezone.now()) .order_by(delete_at, pk) .values_list(pk, flatTrue)[:batch_size], ) if not due_pks: return 0 queryset cls.objects.filter(pk__indue_pks) if cls.delete_after_final_statuses: queryset queryset.filter(status__incls.delete_after_final_statuses) count 0 expired list(queryset.order_by(delete_at, pk)) for obj in expired: obj.delete() count 1 return countdelete_expired是供调度器周期性调用的清理入口设计上对性能做了明确优化回填缺失的 delete_at可选backfill_missingTrue时先通过missing_delete_at_candidates()默认返回空查询集子类覆写找出缺少delete_at的候选行逐条尝试set_delete_at_from_config()回填。missing_delete_at_candidates的子类实现例如Crawlfilter(delete_at__isnullTrue, config__has_keyDELETE_AFTER)archivebox/crawls/models.pySnapshotfilter(Q(config__has_keyDELETE_AFTER) | Q(crawl__config__has_keyDELETE_AFTER))archivebox/core/models.pyProcess基于env与machine.config的 JSON 键判断archivebox/machine/models.py。以delete_at为锚的到期扫描直接对delete_at__ltenow且非 NULL 的行按(delete_at, pk)排序、按batch_size分批取主键。源码注释特别强调在大表上先按 delete_at 索引收窄再叠加status过滤避免扫描数百万行终态数据时先走热门的 status 索引终态过滤若子类声明了delete_after_final_statuses则只删除状态命中的行。例如Snapshot.delete_after_final_statuses (StatusChoices.SEALED,)archivebox/core/models.pyCrawl同样只清理 SEALED 行archivebox/crawls/models.py而ArchiveResult.delete_after_final_statuses FINAL_STATESarchivebox/core/models.py覆盖 succeeded/failed/skipped/noresults 四种终态逐条 delete 计数逐行调用obj.delete()这样 Django 的 pre_delete/post_delete 信号与级联删除仍生效返回删除条数。调度器中的调用链delete_expired由 ArchiveBox 后台 runner 的调度主循环周期性驱动见 archivebox/services/runner.py 与 archivebox/services/runner.py# 紧循环仅做已回填 delete_at的到期清理锚定索引列 if crawl_id is None and now_monotonic - last_retention_at (60.0 if daemon else 1.0): for model in (ArchiveResult, Snapshot, Crawl, Process): model.delete_expired(batch_size100, backfill_missingFalse) last_retention_at now_monotonic # 空闲维护点负责回填缺失的 delete_at需要读 config JSON成本较高 if crawl_id is None and now_monotonic - last_retention_repair_at (60.0 if daemon else 0.0): for model in (ArchiveResult, Snapshot, Crawl, Process): model.delete_expired(batch_size100, backfill_missingTrue) last_retention_repair_at now_monotonic关键设计取舍高频的到期扫描只依赖带索引的delete_at列保证调度紧循环不被拖慢而需要读取 config JSON 才能解析保留期的回填修复被放到本轮没有可运行任务的空闲维护块中执行守护进程模式下每 60 秒一次。源码注释还提到插件结果写入的热路径plugin-result hot path会故意在保存ArchiveResult时不设delete_at由这里的回填统一补齐从而避免每次 hook 事件都加载父 Snapshot/Crawl 配置。ModelWithOutputDir磁盘输出目录的创建与删除ModelWithOutputDir继承ModelWithUUID把数据库行与磁盘输出目录绑定在一起见 archivebox/base_models/models.py。ArchiveBox 的 Snapshot、Crawl、ArchiveResult 都继承它意味着每条记录都对应一个存放归档产物的物理目录。目录命名规则property def output_dir_parent(self) - str: return f{self._meta.model_name}s property def output_dir_name(self) - str: return str(self.id) property def output_dir_str(self) - str: return f{self.output_dir_parent}/{self.output_dir_name} property def output_dir(self) - Path: raise NotImplementedError(f{self.__class__.__name__} must implement output_dir property)output_dir_parent由 Django 元信息中的model_name加复数s推导例如snapshots、crawls、archiveresultsoutput_dir_name就是该行的 UUID 主键字符串output_dir_str形如snapshots/0192a3b4...的相对路径output_dir抽象属性基类直接抛NotImplementedError由子类结合fs_version等字段实现为真实的pathlib.Path。从源码结构看Snapshot.output_dir是通过cached_property依据fs_version与get_storage_path_for_version()计算的见 archivebox/core/models.py 附近注释。save 钩子提交后建目录def save(self, *args, **kwargs): super().save(*args, **kwargs) output_dir Path(self.output_dir) # Avoid holding SQLite write transactions open across slow filesystem work. transaction.on_commit(lambda: output_dir.mkdir(parentsTrue, exist_okTrue)) # Note: index.json is deprecated, models should use write_index_jsonl() for full data实现精妙之处在于使用transaction.on_commit(...)目录创建被推迟到数据库事务提交之后避免 SQLite 写事务在慢速文件系统操作期间长期持有锁。同时源码注释说明历史遗留的index.json已弃用完整数据输出应改用write_index_jsonl()。删除路径安全校验与清理def output_paths_for_delete(self) - tuple[Path, ...]: return (Path(self.output_dir),) classmethod def validate_output_paths_for_delete(cls, paths) - tuple[Path, ...]: data_dir CONSTANTS.DATA_DIR.resolve() safe_paths [] for raw_path in paths: path Path(raw_path) is_safe False for candidate in (path.absolute(), path.resolve()): try: candidate.relative_to(data_dir) is_safe True break except ValueError: continue if not is_safe: raise ValueError(fRefusing to delete output path outside DATA_DIR: {path}) safe_paths.append(path) return tuple(safe_paths) classmethod def delete_output_paths(cls, paths) - None: for path in cls.validate_output_paths_for_delete(paths): if path.is_symlink() or path.is_file(): path.unlink(missing_okTrue) elif path.is_dir(): shutil.rmtree(path, ignore_errorsTrue)安全设计是这里的重中之重validate_output_paths_for_delete强制路径必须位于CONSTANTS.DATA_DIR之内对每个路径同时检查absolute()与resolve()resolve 会展开符号链接二者任一落在DATA_DIR内才放行否则抛出ValueError(Refusing to delete output path outside DATA_DIR: ...)从源头杜绝误删数据目录之外的文件delete_output_paths按类型清理符号链接与普通文件用unlink(missing_okTrue)目录用shutil.rmtree(ignore_errorsTrue)。pre_delete 信号行删除时自动清理磁盘def schedule_delete_cleanup(self, *, using: str | None None) - None: Capture output paths before DB deletion and remove them after commit. paths self.validate_output_paths_for_delete(self.output_paths_for_delete()) transaction.on_commit(lambda: self.delete_output_paths(paths), usingusing) classmethod def register_delete_signal(cls) - None: if cls._delete_signal_registered: return def schedule_output_dir_cleanup(sender, instance, using, **kwargs): if not isinstance(instance, ModelWithOutputDir): return instance.schedule_delete_cleanup(usingusing) pre_delete.connect( schedule_output_dir_cleanup, dispatch_uidarchivebox.output_dir_cleanup_on_delete, weakFalse, ) cls._delete_signal_registered Trueschedule_delete_cleanup在数据库行被删除之前捕获其输出路径并用transaction.on_commit把磁盘清理推迟到删除事务提交成功后执行实现先删库、后删盘的一致性register_delete_signal幂等注册pre_delete信号_delete_signal_registered标志位防止重复连接信号处理器按dispatch_uidarchivebox.output_dir_cleanup_on_delete标识该注册发生在 Django 就绪阶段archivebox/core/apps.py的ready()中调用ModelWithOutputDir.register_delete_signal()见 archivebox/core/apps.py。由于ModelWithDeleteAfter.delete_expired()内部逐条调用obj.delete()被删除的过期记录同样会触发该信号从而把到期行删除与归档产物磁盘清理串联起来——这也是test_config_DELETE_AFTER.pyarchivebox/tests/test_config_DELETE_AFTER.py等测试覆盖的端到端行为。Mixin 组合从模块到业务模型该模块的价值最终体现在具体业务模型的多重继承组合上。以三个核心模型为例# Snapshot五合一archivebox/core/models.py#L538 class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWithNotes, ModelWithHealthStats, ModelWithQueue): ... class Meta( ModelWithDeleteAfter.Meta, ModelWithOutputDir.Meta, ModelWithConfig.Meta, ModelWithNotes.Meta, ModelWithHealthStats.Meta, ModelWithQueue.Meta, ): ...# Crawlarchivebox/crawls/models.py#L150 class Crawl(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWithHealthStats, ModelWithQueue): ... # ArchiveResultarchivebox/core/models.py#L3761 class ArchiveResult(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithNotes): ...组合映射如下Mixin提供的能力主要消费者ModelWithUUIDUUIDv7 主键、created_at/modified_at/created_by、admin/API URL所有核心模型ModelWithNotes备注文本notesSnapshot、ArchiveResult、Crawl、CrawlScheduleModelWithHealthStats成功/失败计数与health百分比Snapshot、Crawl、Machine、NetworkInterface、BinaryModelWithConfigJSON 配置字段与值规范化Snapshot、Crawl、PersonaModelWithDeleteAfterDELETE_AFTER 保留策略、delete_expired批量清理Snapshot、Crawl、ArchiveResult、ProcessModelWithOutputDir输出目录创建、删除安全校验、pre_delete 信号清理Snapshot、Crawl、ArchiveResult每个模型按职责裁剪 Mixin比如Tag只需要ModelWithUUIDPersona只需要ModelWithConfig而快照这种既有配置又有产物还要自动过期的实体则把五个 Mixin 全部继承。Meta也通过多重继承合并各个抽象 Meta保证字段、索引与元信息一致例如 archivebox/core/models.py。小结这套基础模型层的设计要点回顾archivebox/base_models/models.py的完整实现可以提炼出四个值得借鉴的设计原则能力拆分为最小 Mixin六个抽象 Mixin 各自只承担一个横切关注点身份、备注、健康、配置、保留、磁盘业务模型通过多重继承按需组合避免单一大基类导致的字段冗余UUIDv7 主键贯穿全局通过CompactUUIDFielduuid7兼容层archivebox/uuid_compat.py实现时间有序、可离线生成的分布式友好主键事务边界与文件系统解耦无论是建目录save中on_commit还是删目录pre_delete信号 on_commit都把慢速磁盘操作推迟到数据库提交之后且删除路径强制限定在DATA_DIR内配置驱动的自动清理DELETE_AFTER的解析parse_delete_after、行级覆盖get_delete_after_config_value子类化、回填与批量删除delete_expired形成完整闭环并由 runner 调度器在索引友好的前提下分阶段执行。对于希望为自建 Django 项目建立统一模型基建的开发者这套代码是一个结构清晰、注释详尽的现成范本直接复用 archivebox/base_models/models.py 中的 Mixin 组合方式即可快速获得一套具备 UUID 主键、审计字段、健康统计、配置规范化和安全磁盘管理的 ORM 层。【免费下载链接】ArchiveBox Open source self-hosted web archiving. Takes URLs/browser history/bookmarks/Pocket/Pinboard/etc., saves HTML, JS, PDFs, media, and more...项目地址: https://gitcode.com/gh_mirrors/ar/ArchiveBox创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 ArchiveBox 基础模型层解析：基于 UUIDv7 主键与可复用 Django Mixin 的设计实践
ArchiveBox 基础模型层解析基于 UUIDv7 主键与可复用 Django Mixin 的设计实践【免费下载链接】ArchiveBox Open source self-hosted web archiving. Takes URLs/browser history/bookmarks/Pocket/Pinboard/etc., saves HTML, JS, PDFs, media, and more...项目地址: https://gitcode.com/gh_mirrors/ar/ArchiveBox本文围绕archivebox/base_models/models.py展开这是 ArchiveBox 全项目 Django ORM 模型的公共地基。它定义了ModelWithUUID、ModelWithNotes、ModelWithHealthStats、ModelWithConfig、ModelWithDeleteAfter、ModelWithOutputDir六个抽象 Mixin 与AutoDateTimeField字段外加两个模块级工具函数为 Snapshot、Crawl、ArchiveResult、Process 等核心模型统一提供 UUIDv7 主键、审计时间戳、健康统计、JSON 配置、自动清理与磁盘输出目录管理能力。读完本文你将掌握该模块的每个类与函数的实现细节、它们如何被组合进具体业务模型以及如何在自己的 Django 项目中复刻这套模型基建。模块定位全项目模型的公共地基archivebox/base_models/models.py的文件头注释一句话点明其使命Base models using UUIDv7 for all id fields所有 id 字段使用 UUIDv7 的基础模型。该模块位于 archivebox/base_models/models.py被多个 Django app 引用core应用Snapshot、ArchiveResult、Tag等核心归档模型crawls应用Crawl、CrawlSchedulemachine应用Machine、NetworkInterface、Binary、Processpersonas应用Persona。从模块目录看它对外暴露的公共 API 由两部分组成函数2 个normalize_config_json_values(config) - Anyget_or_create_system_user_pk(usernamesystem)类7 个AutoDateTimeField继承django.db.models.DateTimeFieldModelWithUUID继承django.db.models.ModelModelWithNotesModelWithHealthStatsModelWithConfigModelWithDeleteAfterModelWithOutputDir继承ModelWithUUID除AutoDateTimeField外其余六个类均为Meta.abstract True的抽象模型作为 Mixin 被业务模型多重继承组合使用。这种一个字段能力一个 Mixin的拆分方式使得不同业务模型可以按需拼装能力避免重复代码。ModelWithUUIDUUIDv7 主键模型基类ModelWithUUID是模块中最核心的基类为所有继承它的模型提供统一的身份与审计字段。其定义在 archivebox/base_models/models.pyclass ModelWithUUID(models.Model): id CompactUUIDField(primary_keyTrue, defaultuuid7, editableFalse, uniqueTrue) created_at models.DateTimeField(defaulttimezone.now, db_indexTrue) modified_at models.DateTimeField(auto_nowTrue) created_by models.ForeignKey( settings.AUTH_USER_MODEL, on_deletemodels.CASCADE, defaultget_or_create_system_user_pk, nullFalse, db_indexTrue, ) class Meta(TypedModelMeta): abstract True def __str__(self) - str: return f[{self.id}] {self.__class__.__name__}UUIDv7 主键时间有序、无中心协调id字段使用CompactUUIDField(primary_keyTrue, defaultuuid7, editableFalse, uniqueTrue)。其中uuid7是模块内定义的主键默认值生成器见下文UUID7 兼容层CompactUUIDField继承自 Django 的models.UUIDField重写了to_python、from_db_value与deconstruct见 archivebox/uuid_compat.py将 UUID 以无连字符的紧凑十六进制字符串形式存储与展示。选择 UUIDv7 而不是数据库自增主键或随机 UUIDv4关键在于UUIDv7 前缀包含时间戳天然按创建时间排序。这带来两个直接收益批量插入友好UUIDv7 大致随时间单调递增比 UUIDv4 的随机主键更有利于索引局部性与 B 树页填充无需中央协调可以安全地在采集器、后台进程、CLI 等多进程环境中离线生成主键不会发生自增主键常见的竞态。配合created_at默认timezone.now带db_index与modified_atauto_nowTrue每次 save 自动更新每个模型天然拥有完整的时间审计信息。created_by系统操作用户的兜底方案created_by外键指向settings.AUTH_USER_MODELon_deletemodels.CASCADE其默认值由模块级函数get_or_create_system_user_pk提供。该函数在 archivebox/base_models/models.pydef get_or_create_system_user_pk(usernamesystem): User get_user_model() # If theres exactly one superuser, use that for all system operations if User.objects.filter(is_superuserTrue).count() 1: return User.objects.filter(is_superuserTrue).values_list(pk, flatTrue)[0] # Otherwise get or create the system user user, _ User.objects.get_or_create( usernameusername, defaults{is_staff: True, is_superuser: True, email: , password: !}, ) return user.pk其设计意图非常明确唯一超级用户优先若系统中恰好只有一个 superuser多数单机自托管部署的典型形态所有系统自动创建的行如爬虫调度的 Crawl、后台进程写入的 Process都归到该用户名下避免额外创建系统账号否则兜底创建system用户当超级用户数量不为 1多个或零个时get_or_create一个用户名为system的账号is_staffTrue、is_superuserTrue、密码为无效的!并返回其主键。这一设计解决了自动任务产生的行记录由谁创建的问题created_by永远不会为 NULLnullFalse同时又能避免批量创建无意义的用户记录。从源码结构看personas/models.py与crawls/models.py也直接引用了该函数作为外键默认值。便捷 URL 属性ModelWithUUID还提供三个只读属性方便在后台管理与 API 场景快速构造链接见 archivebox/base_models/models.pyproperty def admin_change_url(self) - str: return f/admin/{self._meta.app_label}/{self._meta.model_name}/{self.pk}/change/ property def api_url(self) - str: return str(reverse_lazy(api-1:get_any, args[self.id])) property def api_docs_url(self) - str: return f/api/v1/docs#/{self._meta.app_label.title()}%20Models/api_v1_{self._meta.app_label}_get_{self._meta.db_table}admin_change_url通过 Django admin 的app_label、model_name与主键拼出后台编辑页地址api_url利用命名路由api-1:get_any反解出 v1 API 的通用对象端点ArchiveBox 的 v1 API 支持按任意模型 ID 查询资源参见 archivebox/api/urls.pyapi_docs_url指向 Swagger/OpenAPI 文档中该模型的条目锚点。这三个属性让模板与序列化器无需关心路由细节即可输出标准链接。AutoDateTimeField旧版自动时间戳兼容字段AutoDateTimeField继承django.db.models.DateTimeField其 docstring 明确说明用途DateTimeField that automatically updates on save (legacy compatibility)即为旧版数据提供兼容性的自动更新时间字段。实现位于 archivebox/base_models/models.pyclass AutoDateTimeField(models.DateTimeField): DateTimeField that automatically updates on save (legacy compatibility). def pre_save(self, model_instance, add): if add or self.attname not in model_instance.__dict__ or not model_instance.__dict__[self.attname]: value timezone.now() setattr(model_instance, self.attname, value) return value return super().pre_save(model_instance, add)pre_save是 Django 字段在Model.save()之前调用的钩子。该字段的行为逻辑是若正在新增记录addTrue、或实例中尚无该字段值、或该值为空则强制填充当前时间timezone.now()否则走父类默认逻辑保留已存在的值。与ModelWithUUID.modified_atauto_nowTrue的区别在于AutoDateTimeField允许在特定场景下保留旧值主要用于历史数据迁移与旧版数据模型的兼容读取。模块级函数normalize_config_json_valuesnormalize_config_json_values(config)是ModelWithConfig.save()的核心依赖负责清洗 JSON 配置中的字符串值。实现见 archivebox/base_models/models.pydef normalize_config_json_values(config: Any) - Any: if not isinstance(config, dict): return config normalized dict(config) for key, value in list(normalized.items()): if not isinstance(value, str) or len(value) 2: continue if value[:1] ! or value[-1:] ! : continue try: decoded json.loads(value) except ValueError: continue if isinstance(decoded, str): normalized[key] decoded return normalized它的工作方式非 dict 值直接原样返回遍历每个键值对只处理长度 ≥ 2、以双引号开头且以双引号结尾的字符串尝试用json.loads解析这段引号包裹的字符串若解析成功且结果为字符串则把该值替换为解包后的纯字符串。典型场景当配置值经过 JSON 序列化-反序列化往返后字符串可能被二次编码成\foo\这样的嵌套引号形式。normalize_config_json_values负责把这些字符串化的字符串还原为干净的值保证configJSON 字段中的数据始终规整、可预测。从源码结构看该函数也被 archivebox/machine/models.py 直接导入使用。ModelWithNotes备注字段 MixinModelWithNotes是最轻量的 Mixin只提供一个字段见 archivebox/base_models/models.pyclass ModelWithNotes(models.Model): Mixin for models with a notes field. notes models.TextField(blankTrue, nullFalse, default) class Meta(TypedModelMeta): abstract Truenotes是允许为空blankTrue但非 NULLnullFalse、默认的TextField用于记录用户备注或系统说明。它被Snapshot、ArchiveResult、Crawl、CrawlSchedule等模型继承例如 archivebox/crawls/models.py 中class CrawlSchedule(ModelWithUUID, ModelWithNotes)。ModelWithHealthStats健康统计与原子计数ModelWithHealthStats为模型提供失败次数 / 成功次数两个统计字段与一个健康度百分比计算属性见 archivebox/base_models/models.pyclass ModelWithHealthStats(models.Model): Mixin for models with health tracking fields. num_uses_failed models.PositiveIntegerField(default0) num_uses_succeeded models.PositiveIntegerField(default0) class Meta(TypedModelMeta): abstract True property def admin_change_url(self) - str: return f/admin/{self._meta.app_label}/{self._meta.model_name}/{self.pk}/change/ property def health(self) - int: total max(self.num_uses_failed self.num_uses_succeeded, 1) return round((self.num_uses_succeeded / total) * 100) def increment_health_stats(self, success: bool): Atomically increment success or failure counter using F() expression. field num_uses_succeeded if success else num_uses_failed type(self).objects.filter(pkself.pk).update( **{ field: F(field) 1, modified_at: timezone.now(), }, )值得注意的实现细节health属性返回0~100的整数百分比num_uses_succeeded / (失败 成功)分母通过max(..., 1)兜底避免除零increment_health_stats(success)使用F()表达式原子递增。F(field) 1会把递增操作下推为 SQL 层的SET num_uses_succeeded num_uses_succeeded 1避免读-改-写三步在并发下丢失更新同时顺手刷新modified_at。这是 Django 中做并发安全计数器的标准做法该 Mixin 同时重定义了admin_change_url因此任何同时继承ModelWithUUID与ModelWithHealthStats的模型如Machine、Binary都会得到一致的属性。在业务模型中的实际使用者包括 archivebox/machine/models.py 的Machine、NetworkInterface、Binary以及Snapshot、Crawl等。对于机器、二进制依赖等会被反复调用且可能失败的实体健康统计可以直观反映其可用性。ModelWithConfigJSON 配置字段与写入规范化ModelWithConfig为模型挂载一个 JSON 配置字段并在save()时自动做值规范化见 archivebox/base_models/models.pyclass ModelWithConfig(models.Model): Mixin for models with a JSON config field. config models.JSONField(defaultdict, nullTrue, blankTrue, editableTrue) class Meta(TypedModelMeta): abstract True def save(self, *args, **kwargs): normalized_config normalize_config_json_values(self.config) if normalized_config ! self.config: self.config normalized_config update_fields kwargs.get(update_fields) if update_fields is not None: kwargs[update_fields] tuple(dict.fromkeys([*update_fields, config])) super().save(*args, **kwargs)要点config为 Django 原生JSONField默认dict允许 NULL 与空值可在 admin 中直接编辑save()覆写写入前调用normalize_config_json_values清理嵌套引号字符串若发生了规范化且调用方通过update_fields指定了部分更新字段则用dict.fromkeys去重后把config追加进update_fields确保规范化结果被真正落库这一机制保证了配置在两次读写之间保持幂等与稳定避免存进去是value读出来是value的脏数据。实际使用者包括Snapshot、Crawlarchivebox/crawls/models.py以及Personaarchivebox/personas/models.py 中class Persona(ModelWithConfig)。其中 Snapshot/Crawl 的config承担着冻结本次爬取的配置快照职责ArchiveBox 通过迁移 0018freeze_crawl_config_snapshots将调度时的配置固化到每行保证历史快照可复现见 archivebox/crawls/migrations/0018_freeze_crawl_config_snapshots.py。ModelWithDeleteAfter基于 DELETE_AFTER 的自动过期删除ModelWithDeleteAfter是模块中最具业务复杂度的 Mixin它实现了 ArchiveBox 的保留策略retention policy配置DELETE_AFTER之后到期记录会被自动清理。完整实现见 archivebox/base_models/models.py。字段与类属性class ModelWithDeleteAfter(models.Model): delete_after_final_statuses: tuple[str, ...] () delete_at models.DateTimeField(defaultNone, nullTrue, blankTrue, db_indexTrue)delete_after_final_statuses类级属性声明哪些终态final status记录才允许被删除。默认空元组表示不限制子类覆盖此值来限定范围见下文delete_atDateTimeField可为 NULL带db_index是过期删除的判据。save 钩子与配置解析def save(self, *args, **kwargs): update_fields kwargs.get(update_fields) if self.delete_at is None: self.set_delete_at_from_config() if self.delete_at is not None and update_fields is not None: kwargs[update_fields] tuple(dict.fromkeys([*update_fields, delete_at])) super().save(*args, **kwargs) def get_delete_after_config_value(self): from archivebox.config.common import get_config return get_config(include_machineFalse, resolve_pluginsFalse).DELETE_AFTER def set_delete_at_from_config(self, config_valueNone) - bool: if self.delete_at is not None: return False from archivebox.config.common import parse_delete_after duration parse_delete_after(self.get_delete_after_config_value() if config_value is None else config_value) if duration is None: return False self.delete_at (self.created_at or timezone.now()) duration return True核心逻辑save()覆写当delete_at为空时自动调用set_delete_at_from_config()依据全局配置计算删除时间点同样会感知update_fields并把delete_at加入部分更新集合get_delete_after_config_value()读取全局配置中的DELETE_AFTER。基类默认直接取get_config(...).DELETE_AFTER子类如Snapshot、Crawl、Process会覆写它改为从自身config或父级如 Snapshot 的所属 Crawl的配置中解析实现逐行/逐级可覆盖的保留策略。例如 archivebox/core/models.py 中Snapshot.get_delete_after_config_value调用了resolve_delete_after_config_value(self.config, self.crawl.config)将 Snapshot 自身配置与所属 Crawl 配置合并解析set_delete_at_from_config()把时长解析为timedelta叠加在created_at无则用当前时间上得到delete_at返回是否成功设置。DELETE_AFTER 时长格式时长解析函数parse_delete_after定义在 archivebox/config/common.py它是整个保留策略的语法入口取值含义示例0//none/false/no/off禁用自动删除DELETE_AFTER0h/hr/hrs/hour/hours小时DELETE_AFTER2hd/day/days天DELETE_AFTER7dw/week/weeks周DELETE_AFTER4wmo/month/months月按 30 天计DELETE_AFTER6moy/yr/yrs/year/years年按 365 天计DELETE_AFTER1y规则细节语法为(\d)\s*(单位)的正则全匹配非法格式直接抛ValueError非零时长最短为 1 小时duration timedelta(hours1)会报错该配置在ArchiveConfig中的定义见 archivebox/config/common.py字段校验器validate_delete_after会在配置加载阶段提前做合法性校验。delete_expired批量过期清理classmethod def delete_expired(cls, *, batch_size: int 100, backfill_missing: bool True) - int: if backfill_missing: missing_delete_at list(cls.missing_delete_at_candidates().order_by(created_at, pk)[:batch_size]) for obj in missing_delete_at: if obj.set_delete_at_from_config(): cls.objects.filter(pkobj.pk, delete_at__isnullTrue).update( delete_atobj.delete_at, modified_attimezone.now(), ) # Keep the expiration sweep anchored on delete_at. Some large tables # have millions of final-status rows but almost no retained rows; ... due_pks list( cls.objects.filter(delete_at__isnullFalse, delete_at__ltetimezone.now()) .order_by(delete_at, pk) .values_list(pk, flatTrue)[:batch_size], ) if not due_pks: return 0 queryset cls.objects.filter(pk__indue_pks) if cls.delete_after_final_statuses: queryset queryset.filter(status__incls.delete_after_final_statuses) count 0 expired list(queryset.order_by(delete_at, pk)) for obj in expired: obj.delete() count 1 return countdelete_expired是供调度器周期性调用的清理入口设计上对性能做了明确优化回填缺失的 delete_at可选backfill_missingTrue时先通过missing_delete_at_candidates()默认返回空查询集子类覆写找出缺少delete_at的候选行逐条尝试set_delete_at_from_config()回填。missing_delete_at_candidates的子类实现例如Crawlfilter(delete_at__isnullTrue, config__has_keyDELETE_AFTER)archivebox/crawls/models.pySnapshotfilter(Q(config__has_keyDELETE_AFTER) | Q(crawl__config__has_keyDELETE_AFTER))archivebox/core/models.pyProcess基于env与machine.config的 JSON 键判断archivebox/machine/models.py。以delete_at为锚的到期扫描直接对delete_at__ltenow且非 NULL 的行按(delete_at, pk)排序、按batch_size分批取主键。源码注释特别强调在大表上先按 delete_at 索引收窄再叠加status过滤避免扫描数百万行终态数据时先走热门的 status 索引终态过滤若子类声明了delete_after_final_statuses则只删除状态命中的行。例如Snapshot.delete_after_final_statuses (StatusChoices.SEALED,)archivebox/core/models.pyCrawl同样只清理 SEALED 行archivebox/crawls/models.py而ArchiveResult.delete_after_final_statuses FINAL_STATESarchivebox/core/models.py覆盖 succeeded/failed/skipped/noresults 四种终态逐条 delete 计数逐行调用obj.delete()这样 Django 的 pre_delete/post_delete 信号与级联删除仍生效返回删除条数。调度器中的调用链delete_expired由 ArchiveBox 后台 runner 的调度主循环周期性驱动见 archivebox/services/runner.py 与 archivebox/services/runner.py# 紧循环仅做已回填 delete_at的到期清理锚定索引列 if crawl_id is None and now_monotonic - last_retention_at (60.0 if daemon else 1.0): for model in (ArchiveResult, Snapshot, Crawl, Process): model.delete_expired(batch_size100, backfill_missingFalse) last_retention_at now_monotonic # 空闲维护点负责回填缺失的 delete_at需要读 config JSON成本较高 if crawl_id is None and now_monotonic - last_retention_repair_at (60.0 if daemon else 0.0): for model in (ArchiveResult, Snapshot, Crawl, Process): model.delete_expired(batch_size100, backfill_missingTrue) last_retention_repair_at now_monotonic关键设计取舍高频的到期扫描只依赖带索引的delete_at列保证调度紧循环不被拖慢而需要读取 config JSON 才能解析保留期的回填修复被放到本轮没有可运行任务的空闲维护块中执行守护进程模式下每 60 秒一次。源码注释还提到插件结果写入的热路径plugin-result hot path会故意在保存ArchiveResult时不设delete_at由这里的回填统一补齐从而避免每次 hook 事件都加载父 Snapshot/Crawl 配置。ModelWithOutputDir磁盘输出目录的创建与删除ModelWithOutputDir继承ModelWithUUID把数据库行与磁盘输出目录绑定在一起见 archivebox/base_models/models.py。ArchiveBox 的 Snapshot、Crawl、ArchiveResult 都继承它意味着每条记录都对应一个存放归档产物的物理目录。目录命名规则property def output_dir_parent(self) - str: return f{self._meta.model_name}s property def output_dir_name(self) - str: return str(self.id) property def output_dir_str(self) - str: return f{self.output_dir_parent}/{self.output_dir_name} property def output_dir(self) - Path: raise NotImplementedError(f{self.__class__.__name__} must implement output_dir property)output_dir_parent由 Django 元信息中的model_name加复数s推导例如snapshots、crawls、archiveresultsoutput_dir_name就是该行的 UUID 主键字符串output_dir_str形如snapshots/0192a3b4...的相对路径output_dir抽象属性基类直接抛NotImplementedError由子类结合fs_version等字段实现为真实的pathlib.Path。从源码结构看Snapshot.output_dir是通过cached_property依据fs_version与get_storage_path_for_version()计算的见 archivebox/core/models.py 附近注释。save 钩子提交后建目录def save(self, *args, **kwargs): super().save(*args, **kwargs) output_dir Path(self.output_dir) # Avoid holding SQLite write transactions open across slow filesystem work. transaction.on_commit(lambda: output_dir.mkdir(parentsTrue, exist_okTrue)) # Note: index.json is deprecated, models should use write_index_jsonl() for full data实现精妙之处在于使用transaction.on_commit(...)目录创建被推迟到数据库事务提交之后避免 SQLite 写事务在慢速文件系统操作期间长期持有锁。同时源码注释说明历史遗留的index.json已弃用完整数据输出应改用write_index_jsonl()。删除路径安全校验与清理def output_paths_for_delete(self) - tuple[Path, ...]: return (Path(self.output_dir),) classmethod def validate_output_paths_for_delete(cls, paths) - tuple[Path, ...]: data_dir CONSTANTS.DATA_DIR.resolve() safe_paths [] for raw_path in paths: path Path(raw_path) is_safe False for candidate in (path.absolute(), path.resolve()): try: candidate.relative_to(data_dir) is_safe True break except ValueError: continue if not is_safe: raise ValueError(fRefusing to delete output path outside DATA_DIR: {path}) safe_paths.append(path) return tuple(safe_paths) classmethod def delete_output_paths(cls, paths) - None: for path in cls.validate_output_paths_for_delete(paths): if path.is_symlink() or path.is_file(): path.unlink(missing_okTrue) elif path.is_dir(): shutil.rmtree(path, ignore_errorsTrue)安全设计是这里的重中之重validate_output_paths_for_delete强制路径必须位于CONSTANTS.DATA_DIR之内对每个路径同时检查absolute()与resolve()resolve 会展开符号链接二者任一落在DATA_DIR内才放行否则抛出ValueError(Refusing to delete output path outside DATA_DIR: ...)从源头杜绝误删数据目录之外的文件delete_output_paths按类型清理符号链接与普通文件用unlink(missing_okTrue)目录用shutil.rmtree(ignore_errorsTrue)。pre_delete 信号行删除时自动清理磁盘def schedule_delete_cleanup(self, *, using: str | None None) - None: Capture output paths before DB deletion and remove them after commit. paths self.validate_output_paths_for_delete(self.output_paths_for_delete()) transaction.on_commit(lambda: self.delete_output_paths(paths), usingusing) classmethod def register_delete_signal(cls) - None: if cls._delete_signal_registered: return def schedule_output_dir_cleanup(sender, instance, using, **kwargs): if not isinstance(instance, ModelWithOutputDir): return instance.schedule_delete_cleanup(usingusing) pre_delete.connect( schedule_output_dir_cleanup, dispatch_uidarchivebox.output_dir_cleanup_on_delete, weakFalse, ) cls._delete_signal_registered Trueschedule_delete_cleanup在数据库行被删除之前捕获其输出路径并用transaction.on_commit把磁盘清理推迟到删除事务提交成功后执行实现先删库、后删盘的一致性register_delete_signal幂等注册pre_delete信号_delete_signal_registered标志位防止重复连接信号处理器按dispatch_uidarchivebox.output_dir_cleanup_on_delete标识该注册发生在 Django 就绪阶段archivebox/core/apps.py的ready()中调用ModelWithOutputDir.register_delete_signal()见 archivebox/core/apps.py。由于ModelWithDeleteAfter.delete_expired()内部逐条调用obj.delete()被删除的过期记录同样会触发该信号从而把到期行删除与归档产物磁盘清理串联起来——这也是test_config_DELETE_AFTER.pyarchivebox/tests/test_config_DELETE_AFTER.py等测试覆盖的端到端行为。Mixin 组合从模块到业务模型该模块的价值最终体现在具体业务模型的多重继承组合上。以三个核心模型为例# Snapshot五合一archivebox/core/models.py#L538 class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWithNotes, ModelWithHealthStats, ModelWithQueue): ... class Meta( ModelWithDeleteAfter.Meta, ModelWithOutputDir.Meta, ModelWithConfig.Meta, ModelWithNotes.Meta, ModelWithHealthStats.Meta, ModelWithQueue.Meta, ): ...# Crawlarchivebox/crawls/models.py#L150 class Crawl(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWithHealthStats, ModelWithQueue): ... # ArchiveResultarchivebox/core/models.py#L3761 class ArchiveResult(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithNotes): ...组合映射如下Mixin提供的能力主要消费者ModelWithUUIDUUIDv7 主键、created_at/modified_at/created_by、admin/API URL所有核心模型ModelWithNotes备注文本notesSnapshot、ArchiveResult、Crawl、CrawlScheduleModelWithHealthStats成功/失败计数与health百分比Snapshot、Crawl、Machine、NetworkInterface、BinaryModelWithConfigJSON 配置字段与值规范化Snapshot、Crawl、PersonaModelWithDeleteAfterDELETE_AFTER 保留策略、delete_expired批量清理Snapshot、Crawl、ArchiveResult、ProcessModelWithOutputDir输出目录创建、删除安全校验、pre_delete 信号清理Snapshot、Crawl、ArchiveResult每个模型按职责裁剪 Mixin比如Tag只需要ModelWithUUIDPersona只需要ModelWithConfig而快照这种既有配置又有产物还要自动过期的实体则把五个 Mixin 全部继承。Meta也通过多重继承合并各个抽象 Meta保证字段、索引与元信息一致例如 archivebox/core/models.py。小结这套基础模型层的设计要点回顾archivebox/base_models/models.py的完整实现可以提炼出四个值得借鉴的设计原则能力拆分为最小 Mixin六个抽象 Mixin 各自只承担一个横切关注点身份、备注、健康、配置、保留、磁盘业务模型通过多重继承按需组合避免单一大基类导致的字段冗余UUIDv7 主键贯穿全局通过CompactUUIDFielduuid7兼容层archivebox/uuid_compat.py实现时间有序、可离线生成的分布式友好主键事务边界与文件系统解耦无论是建目录save中on_commit还是删目录pre_delete信号 on_commit都把慢速磁盘操作推迟到数据库提交之后且删除路径强制限定在DATA_DIR内配置驱动的自动清理DELETE_AFTER的解析parse_delete_after、行级覆盖get_delete_after_config_value子类化、回填与批量删除delete_expired形成完整闭环并由 runner 调度器在索引友好的前提下分阶段执行。对于希望为自建 Django 项目建立统一模型基建的开发者这套代码是一个结构清晰、注释详尽的现成范本直接复用 archivebox/base_models/models.py 中的 Mixin 组合方式即可快速获得一套具备 UUID 主键、审计字段、健康统计、配置规范化和安全磁盘管理的 ORM 层。【免费下载链接】ArchiveBox Open source self-hosted web archiving. Takes URLs/browser history/bookmarks/Pocket/Pinboard/etc., saves HTML, JS, PDFs, media, and more...项目地址: https://gitcode.com/gh_mirrors/ar/ArchiveBox创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考