ArchiveBox 基础模型层解析:基于 UUIDv7 主键与可复用 Django Mixin 的设计实践
2026/9/20 22:44:41 网站建设 项目流程

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 模型的公共地基。它定义了ModelWithUUIDModelWithNotesModelWithHealthStatsModelWithConfigModelWithDeleteAfterModelWithOutputDir六个抽象 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应用:SnapshotArchiveResultTag等核心归档模型;
  • crawls应用:CrawlCrawlSchedule
  • machine应用:MachineNetworkInterfaceBinaryProcess
  • personas应用:Persona

从模块目录看,它对外暴露的公共 API 由两部分组成:

函数(2 个)

  • normalize_config_json_values(config) -> Any
  • get_or_create_system_user_pk(username='system')

类(7 个)

  • AutoDateTimeField(继承django.db.models.DateTimeField
  • ModelWithUUID(继承django.db.models.Model
  • ModelWithNotes
  • ModelWithHealthStats
  • ModelWithConfig
  • ModelWithDeleteAfter
  • ModelWithOutputDir(继承ModelWithUUID

AutoDateTimeField外,其余六个类均为Meta.abstract = True的抽象模型,作为 Mixin 被业务模型多重继承组合使用。这种"一个字段能力一个 Mixin"的拆分方式,使得不同业务模型可以按需拼装能力,避免重复代码。

ModelWithUUID:UUIDv7 主键模型基类

ModelWithUUID是模块中最核心的基类,为所有继承它的模型提供统一的身份与审计字段。其定义在 archivebox/base_models/models.py:

class ModelWithUUID(models.Model): id = CompactUUIDField(primary_key=True, default=uuid7, editable=False, unique=True) created_at = models.DateTimeField(default=timezone.now, db_index=True) modified_at = models.DateTimeField(auto_now=True) created_by = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.CASCADE, default=get_or_create_system_user_pk, null=False, db_index=True, ) class Meta(TypedModelMeta): abstract = True def __str__(self) -> str: return f"[{self.id}] {self.__class__.__name__}"

UUIDv7 主键:时间有序、无中心协调

id字段使用CompactUUIDField(primary_key=True, default=uuid7, editable=False, unique=True)。其中:

  • uuid7是模块内定义的主键默认值生成器(见下文"UUID7 兼容层");
  • CompactUUIDField继承自 Django 的models.UUIDField,重写了to_pythonfrom_db_valuedeconstruct(见 archivebox/uuid_compat.py),将 UUID 以无连字符的紧凑十六进制字符串形式存储与展示。

选择 UUIDv7 而不是数据库自增主键或随机 UUIDv4,关键在于UUIDv7 前缀包含时间戳,天然按创建时间排序。这带来两个直接收益:

  1. 批量插入友好:UUIDv7 大致随时间单调递增,比 UUIDv4 的随机主键更有利于索引局部性与 B+ 树页填充;
  2. 无需中央协调:可以安全地在采集器、后台进程、CLI 等多进程环境中离线生成主键,不会发生自增主键常见的竞态。

配合created_at(默认timezone.now,带db_index)与modified_atauto_now=True,每次 save 自动更新),每个模型天然拥有完整的时间审计信息。

created_by:系统操作用户的兜底方案

created_by外键指向settings.AUTH_USER_MODELon_delete=models.CASCADE,其默认值由模块级函数get_or_create_system_user_pk提供。该函数在 archivebox/base_models/models.py:

def get_or_create_system_user_pk(username="system"): User = get_user_model() # If there's exactly one superuser, use that for all system operations if User.objects.filter(is_superuser=True).count() == 1: return User.objects.filter(is_superuser=True).values_list("pk", flat=True)[0] # Otherwise get or create the system user user, _ = User.objects.get_or_create( username=username, defaults={"is_staff": True, "is_superuser": True, "email": "", "password": "!"}, ) return user.pk

其设计意图非常明确:

  • 唯一超级用户优先:若系统中恰好只有一个 superuser(多数单机自托管部署的典型形态),所有系统自动创建的行(如爬虫调度的 Crawl、后台进程写入的 Process)都归到该用户名下,避免额外创建系统账号;
  • 否则兜底创建system用户:当超级用户数量不为 1(多个或零个)时,get_or_create一个用户名为system的账号(is_staff=Trueis_superuser=True、密码为无效的!),并返回其主键。

这一设计解决了"自动任务产生的行记录由谁创建"的问题:created_by永远不会为 NULL(null=False),同时又能避免批量创建无意义的用户记录。从源码结构看,personas/models.pycrawls/models.py也直接引用了该函数作为外键默认值。

便捷 URL 属性

ModelWithUUID还提供三个只读属性,方便在后台管理与 API 场景快速构造链接(见 archivebox/base_models/models.py):

@property 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_labelmodel_name与主键拼出后台编辑页地址;
  • api_url:利用命名路由api-1:get_any反解出 v1 API 的通用对象端点(ArchiveBox 的 v1 API 支持按任意模型 ID 查询资源,参见 archivebox/api/urls.py);
  • api_docs_url:指向 Swagger/OpenAPI 文档中该模型的条目锚点。

这三个属性让模板与序列化器无需关心路由细节即可输出标准链接。

AutoDateTimeField:旧版自动时间戳兼容字段

AutoDateTimeField继承django.db.models.DateTimeField,其 docstring 明确说明用途:"DateTimeField that automatically updates on save (legacy compatibility)",即为旧版数据提供兼容性的自动更新时间字段。实现位于 archivebox/base_models/models.py:

class 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()之前调用的钩子。该字段的行为逻辑是:

  • 若正在新增记录(add=True)、或实例中尚无该字段值、或该值为空,则强制填充当前时间timezone.now()
  • 否则走父类默认逻辑(保留已存在的值)。

ModelWithUUID.modified_atauto_now=True)的区别在于:AutoDateTimeField允许在特定场景下保留旧值,主要用于历史数据迁移与旧版数据模型的兼容读取。

模块级函数:normalize_config_json_values

normalize_config_json_values(config)ModelWithConfig.save()的核心依赖,负责清洗 JSON 配置中的字符串值。实现见 archivebox/base_models/models.py:

def 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

它的工作方式:

  1. 非 dict 值直接原样返回;
  2. 遍历每个键值对,只处理"长度 ≥ 2、以双引号开头且以双引号结尾的字符串";
  3. 尝试用json.loads解析这段引号包裹的字符串;
  4. 若解析成功且结果为字符串,则把该值替换为解包后的纯字符串。

典型场景:当配置值经过 JSON 序列化-反序列化往返后,字符串可能被二次编码成"\"foo\""这样的嵌套引号形式。normalize_config_json_values负责把这些"字符串化的字符串"还原为干净的值,保证configJSON 字段中的数据始终规整、可预测。从源码结构看,该函数也被 archivebox/machine/models.py 直接导入使用。

ModelWithNotes:备注字段 Mixin

ModelWithNotes是最轻量的 Mixin,只提供一个字段(见 archivebox/base_models/models.py):

class ModelWithNotes(models.Model): """Mixin for models with a notes field.""" notes = models.TextField(blank=True, null=False, default="") class Meta(TypedModelMeta): abstract = True

notes是允许为空(blank=True)但非 NULL(null=False、默认"")的TextField,用于记录用户备注或系统说明。它被SnapshotArchiveResultCrawlCrawlSchedule等模型继承,例如 archivebox/crawls/models.py 中class CrawlSchedule(ModelWithUUID, ModelWithNotes)

ModelWithHealthStats:健康统计与原子计数

ModelWithHealthStats为模型提供"失败次数 / 成功次数"两个统计字段与一个健康度百分比计算属性(见 archivebox/base_models/models.py):

class ModelWithHealthStats(models.Model): """Mixin for models with health tracking fields.""" num_uses_failed = models.PositiveIntegerField(default=0) num_uses_succeeded = models.PositiveIntegerField(default=0) 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(pk=self.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,因此任何同时继承ModelWithUUIDModelWithHealthStats的模型(如MachineBinary)都会得到一致的属性。

在业务模型中的实际使用者包括 archivebox/machine/models.py 的MachineNetworkInterfaceBinary,以及SnapshotCrawl等。对于机器、二进制依赖等"会被反复调用且可能失败"的实体,健康统计可以直观反映其可用性。

ModelWithConfig:JSON 配置字段与写入规范化

ModelWithConfig为模型挂载一个 JSON 配置字段,并在save()时自动做值规范化(见 archivebox/base_models/models.py):

class ModelWithConfig(models.Model): """Mixin for models with a JSON config field.""" config = models.JSONField(default=dict, null=True, blank=True, editable=True) 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""的脏数据。

实际使用者包括SnapshotCrawl(archivebox/crawls/models.py)以及Persona(archivebox/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(default=None, null=True, blank=True, db_index=True)
  • 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_machine=False, resolve_plugins=False).DELETE_AFTER def set_delete_at_from_config(self, config_value=None) -> 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;子类(如SnapshotCrawlProcess)会覆写它,改为从自身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_AFTER=0
h/hr/hrs/hour/hours小时DELETE_AFTER=2h
d/day/daysDELETE_AFTER=7d
w/week/weeksDELETE_AFTER=4w
mo/month/months月(按 30 天计)DELETE_AFTER=6mo
y/yr/yrs/year/years年(按 365 天计)DELETE_AFTER=1y

规则细节:

  • 语法为(\d+)\s*(单位)的正则全匹配,非法格式直接抛ValueError
  • 非零时长最短为 1 小时duration < timedelta(hours=1)会报错;
  • 该配置在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(pk=obj.pk, delete_at__isnull=True).update( delete_at=obj.delete_at, modified_at=timezone.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__isnull=False, delete_at__lte=timezone.now()) .order_by("delete_at", "pk") .values_list("pk", flat=True)[:batch_size], ) if not due_pks: return 0 queryset = cls.objects.filter(pk__in=due_pks) if cls.delete_after_final_statuses: queryset = queryset.filter(status__in=cls.delete_after_final_statuses) count = 0 expired = list(queryset.order_by("delete_at", "pk")) for obj in expired: obj.delete() count += 1 return count

delete_expired是供调度器周期性调用的清理入口,设计上对性能做了明确优化:

  1. 回填缺失的 delete_at(可选)backfill_missing=True时,先通过missing_delete_at_candidates()(默认返回空查询集,子类覆写)找出缺少delete_at的候选行,逐条尝试set_delete_at_from_config()回填。missing_delete_at_candidates的子类实现例如:
    • Crawlfilter(delete_at__isnull=True, config__has_key="DELETE_AFTER")(archivebox/crawls/models.py);
    • Snapshotfilter(Q(config__has_key="DELETE_AFTER") | Q(crawl__config__has_key="DELETE_AFTER"))(archivebox/core/models.py);
    • Process:基于envmachine.config的 JSON 键判断(archivebox/machine/models.py)。
  2. delete_at为锚的到期扫描:直接对delete_at__lte=now且非 NULL 的行按(delete_at, pk)排序、按batch_size分批取主键。源码注释特别强调:在大表上先按 delete_at 索引收窄,再叠加status过滤,避免扫描数百万行终态数据时先走热门的 status 索引;
  3. 终态过滤:若子类声明了delete_after_final_statuses,则只删除状态命中的行。例如Snapshot.delete_after_final_statuses = (StatusChoices.SEALED,)(archivebox/core/models.py),Crawl同样只清理 SEALED 行(archivebox/crawls/models.py),而ArchiveResult.delete_after_final_statuses = FINAL_STATES(archivebox/core/models.py),覆盖 succeeded/failed/skipped/noresults 四种终态;
  4. 逐条 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_size=100, backfill_missing=False) 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_size=100, backfill_missing=True) 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推导,例如snapshotscrawlsarchiveresults
  • output_dir_name:就是该行的 UUID 主键字符串;
  • output_dir_str:形如snapshots/0192a3b4...的相对路径;
  • output_dir抽象属性,基类直接抛NotImplementedError,由子类结合fs_version等字段实现为真实的pathlib.Path。从源码结构看,Snapshot.output_dir是通过@cached_property依据fs_versionget_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(parents=True, exist_ok=True)) # 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(f"Refusing 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_ok=True) elif path.is_dir(): shutil.rmtree(path, ignore_errors=True)

安全设计是这里的重中之重:

  • 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_ok=True),目录用shutil.rmtree(ignore_errors=True)

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), using=using) @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(using=using) pre_delete.connect( schedule_output_dir_cleanup, dispatch_uid="archivebox.output_dir_cleanup_on_delete", weak=False, ) cls._delete_signal_registered = True
  • schedule_delete_cleanup:在数据库行被删除之前捕获其输出路径,并用transaction.on_commit把磁盘清理推迟到删除事务提交成功后执行,实现"先删库、后删盘"的一致性;
  • register_delete_signal:幂等注册pre_delete信号(_delete_signal_registered标志位防止重复连接),信号处理器按dispatch_uid="archivebox.output_dir_cleanup_on_delete"标识;
  • 该注册发生在 Django 就绪阶段:archivebox/core/apps.pyready()中调用ModelWithOutputDir.register_delete_signal()(见 archivebox/core/apps.py)。

由于ModelWithDeleteAfter.delete_expired()内部逐条调用obj.delete(),被删除的过期记录同样会触发该信号,从而把"到期行删除"与"归档产物磁盘清理"串联起来——这也是test_config_DELETE_AFTER.py(archivebox/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, ): ...
# Crawl(archivebox/crawls/models.py#L150) class Crawl(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWithHealthStats, ModelWithQueue): ... # ArchiveResult(archivebox/core/models.py#L3761) class ArchiveResult(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithNotes): ...

组合映射如下:

Mixin提供的能力主要消费者
ModelWithUUIDUUIDv7 主键、created_at/modified_at/created_by、admin/API URL所有核心模型
ModelWithNotes备注文本notesSnapshot、ArchiveResult、Crawl、CrawlSchedule
ModelWithHealthStats成功/失败计数与health百分比Snapshot、Crawl、Machine、NetworkInterface、Binary
ModelWithConfigJSON 配置字段与值规范化Snapshot、Crawl、Persona
ModelWithDeleteAfterDELETE_AFTER 保留策略、delete_expired批量清理Snapshot、Crawl、ArchiveResult、Process
ModelWithOutputDir输出目录创建、删除安全校验、pre_delete 信号清理Snapshot、Crawl、ArchiveResult

每个模型按职责裁剪 Mixin:比如Tag只需要ModelWithUUIDPersona只需要ModelWithConfig;而快照这种"既有配置又有产物还要自动过期"的实体则把五个 Mixin 全部继承。Meta也通过多重继承合并各个抽象 Meta,保证字段、索引与元信息一致(例如 archivebox/core/models.py)。

小结:这套基础模型层的设计要点

回顾archivebox/base_models/models.py的完整实现,可以提炼出四个值得借鉴的设计原则:

  1. 能力拆分为最小 Mixin:六个抽象 Mixin 各自只承担一个横切关注点(身份、备注、健康、配置、保留、磁盘),业务模型通过多重继承按需组合,避免单一大基类导致的字段冗余;
  2. UUIDv7 主键贯穿全局:通过CompactUUIDField+uuid7兼容层(archivebox/uuid_compat.py)实现时间有序、可离线生成的分布式友好主键;
  3. 事务边界与文件系统解耦:无论是建目录(saveon_commit)还是删目录(pre_delete信号 +on_commit),都把慢速磁盘操作推迟到数据库提交之后,且删除路径强制限定在DATA_DIR内;
  4. 配置驱动的自动清理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),仅供参考

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询