Add syntax for declarative record descriptors

.github/workflows/dissect-ci.yml

@@ -12,6 +12,8 @@ jobs:
   ci:
     uses: fox-it/dissect-workflow-templates/.github/workflows/dissect-ci-template.yml@main
     secrets: inherit
+    with:
+      run-benchmarks: true
 
   publish:
     if: ${{ github.ref_name == 'main' || github.ref_type == 'tag' }}

README.md

@@ -71,6 +71,59 @@ $ rdump output.records.gz
 <my/record ip=net.ipaddress('8.8.8.8') description='google dns'>
 ```
 
+### Declarative record definitions
+
+Besides the `RecordDescriptor` API shown above, records can also be defined declaratively by subclassing `RecordBase`
+and annotating fields, like a `dataclass` or `NamedTuple`:
+
+```python
+from datetime import datetime
+from typing import Annotated
+
+from flow.record.declarative import RecordBase, field
+
+
+class HttpRequestRecord(RecordBase, name="http/request"):
+    ts: datetime
+    url: str  # native types map (e.g. str -> string)
+    status: Annotated[int, "uint32"]  # or field(typename="uint32")
+    remote: str
+
+
+record = HttpRequestRecord(ts=datetime.now(), url="http://flow.record", status=200, remote="127.0.0.1")
+```
+
+Inheritance works as expected — a subclass extends its parent's fields:
+
+```python
+class HttpResponseRecord(HttpRequestRecord, name="http/response"):
+    body: bytes
+```
+
+Use `field(init=False)` for derived fields and populate them from `InitVar` inputs in a `__post_init__` hook:
+
+```python
+from dataclasses import InitVar
+
+
+class HostRecord(RecordBase, name="example/host"):
+    hostname: str = field(init=False)
+    target: InitVar[object]
+
+    def __post_init__(self, target: object) -> None:
+        self.hostname = target.hostname
+```
+
+> [!NOTE]
+> Field annotations are evaluated at runtime (via `get_type_hints`). If your project uses Ruff's `flake8-type-checking`
+> (`TC`) rules, add `RecordBase` to the runtime-evaluated base classes once, so imports used only in field annotations
+> are not moved into a `TYPE_CHECKING` block:
+>
+> ```toml
+> [tool.ruff.lint.flake8-type-checking]
+> runtime-evaluated-base-classes = ["flow.record.declarative.RecordBase"]
+> ```
+
 ### Selectors
 
 We can also use `selectors` for filtering and selecting records using a query (Python like syntax), e.g.:

flow/record/base.py

@@ -114,6 +114,7 @@ def _unpack(__cls, {args}):
 {unpack_code}
 """
 
+_FIELDTYPES_PREFIX = "flow.record.fieldtypes"
 
 if env_excluded_fields := os.environ.get("FLOW_RECORD_IGNORE"):
     IGNORE_FIELDS_FOR_COMPARISON = set(env_excluded_fields.split(","))
@@ -153,6 +154,11 @@ def _unpack(cls, data: Any) -> Any:
         return data
 
 
+def _default_is_trivial(field_type: type[FieldType]) -> bool:
+    """Return whether ``field_type`` uses the base :meth:`FieldType.default` (i.e. returns ``None``)."""
+    return field_type.default.__func__ is FieldType.default.__func__
+
+
 class Record:
     __slots__ = ()
 
@@ -441,10 +447,7 @@ def _generate_record_class(name: str, fields: tuple[tuple[str, str]]) -> type:
         args = ", ".join([f"{k}=None" for k in all_fields])
         unpack_code = "\t\treturn __cls(\n"
         for field in all_fields.values():
-            if field.type.default == FieldType.default:
-                default = FieldType.default()
-            else:
-                default = f"_field_{field.name}.type.default()"
+            default = FieldType.default() if _default_is_trivial(field.type) else f"_field_{field.name}.type.default()"
             init_code += f"\t\t__self.{field.name} = {field.name} if {field.name} is not None else {default}\n"
             unpack_code += (
                 "\t\t\t{field} = _field_{field}.type._unpack({field}) " + "if {field} is not None else {default},\n"
@@ -949,8 +952,6 @@ def fieldtype(clspath: str) -> FieldType:
     Returns:
         The FieldType class.
     """
-    base_module_path = "flow.record.fieldtypes"
-
     if clspath.endswith("[]"):
         origpath = clspath
         clspath = clspath[:-2]
@@ -962,13 +963,13 @@ def fieldtype(clspath: str) -> FieldType:
         raise AttributeError(f"Invalid field type: {clspath}")
 
     namespace, _, clsname = clspath.rpartition(".")
-    module_path = f"{base_module_path}.{namespace}" if namespace else base_module_path
+    module_path = f"{_FIELDTYPES_PREFIX}.{namespace}" if namespace else _FIELDTYPES_PREFIX
     mod = importlib.import_module(module_path)
 
     fieldtype_cls = getattr(mod, clsname)
 
     if islist:
-        base_mod = importlib.import_module(base_module_path)
+        base_mod = importlib.import_module(_FIELDTYPES_PREFIX)
         listtype = type(origpath, base_mod.typedlist.__bases__, dict(base_mod.typedlist.__dict__))
         listtype.__type__, fieldtype_cls = fieldtype_cls, listtype