Understanding Flowsint's Type System and Pydantic Models

Flowsint models every piece of investigative data—such as domains, IPs, and social accounts—as a Pydantic-based class that inherits from FlowsintType, registers automatically through the @flowsint_type decorator, and sanitizes input via field-level and model-level validators.

The Flowsint type system lives inside the flowsint-types package in the reconurge/flowsint repository. It provides a typed, self-documenting, and graph-ready data model that the core platform, enrichers, and API consume through a lightweight global registry.

The FlowsintType Base Class and Configuration

In flowsint-types/src/flowsint_types/flowsint_base.py, the FlowsintType class extends Pydantic's BaseModel to serve as the foundation for all investigative entities.

class FlowsintType(BaseModel):
    nodeLabel: Optional[str] = Field(
        None,
        description="UI-readable label for this entity, the one used on the graph.",
        title="Label",
    )
    model_config = ConfigDict(extra="allow")   # accept user-supplied keys

All entity subclasses inherit this base, gaining the optional nodeLabel field that the UI and Neo4j graph layer consume. The extra="allow" setting ensures that user-supplied keys pass through validation without raising errors.

Automatic Type Registration with @flowsint_type

The registry mechanism in flowsint-types/src/flowsint_types/registry.py eliminates manual imports by collecting every decorated class at runtime. The @flowsint_type decorator forwards each subclass to TYPE_REGISTRY.register.

def flowsint_type(cls: Type[T]) -> Type[T]:
    return TYPE_REGISTRY.register(cls)

TYPE_REGISTRY maintains two internal dictionaries:

• class_name → class (e.g., "Domain" → Domain)
• lowercase_name → class (e.g., "domain" → Domain) — required for Neo4j compatibility.

At startup, load_all_types() walks the flowsint_types package, imports every module, and triggers each decorator. This guarantees that all types are available for dynamic lookup without explicit imports.

Validation Patterns in Flowsint Pydantic Models

Concrete types in Flowsint combine two Pydantic validation mechanisms to sanitize raw input and compute derived fields.

Field-Level Sanitization with @field_validator

The @field_validator decorator runs before the model is instantiated, allowing conversion of raw values into complex objects. In flowsint-types/src/flowsint_types/domain.py, the Domain class normalizes hostnames before construction. In social_account.py, a validator coerces a plain string into a Username object.

Model-Level Post-Processing with @model_validator

The @model_validator(mode="after") decorator runs after instantiation, enabling classes to compute fields such as nodeLabel or a composite id. Because the model already exists, validators can read sibling fields and mutate state safely.

Concrete Type Examples: Domain and SocialAccount

The individual type modules in flowsint-types/src/flowsint_types/ demonstrate how these patterns work in practice.

Domain Validation and Root Detection

In domain.py, the Domain class defines a primary field, validates hostnames, and detects root domains through chained validators.

class Domain(FlowsintType):
    domain: str = Field(..., json_schema_extra={"primary": True})
    root: Optional[bool] = Field(True)

    @field_validator("domain")
    @classmethod
    def validate_domain(cls, v: str) -> str:
        # normalises and validates the hostname

        ...

    @model_validator(mode="after")
    def check_root(self) -> Self:
        self.root = is_root_domain(self.domain)
        return self

    @model_validator(mode="after")
    def compute_label(self) -> Self:
        self.nodeLabel = self.domain
        return self

The json_schema_extra={"primary": True} marker tells the graph layer which property to use as the primary key for Neo4j storage.

SocialAccount Username Conversion and ID Composition

In social_account.py, the SocialAccount class converts raw username strings into structured Username objects and builds deterministic identifiers.

class SocialAccount(FlowsintType):
    username: Username = Field(...)
    platform: Optional[str] = Field(...)
    id: Optional[str] = Field(..., json_schema_extra={"primary": True})

    @field_validator("username", mode="before")
    @classmethod
    def convert_username(cls, v: Union[str, Username]) -> Username:
        return Username(value=v) if isinstance(v, str) else v

    @model_validator(mode="after")
    def compute_label_and_id(self) -> Self:
        # Build a deterministic id "username@platform" and a friendly label

        if self.username and self.platform:
            self.id = f"{self.username.value}@{self.platform}"
        elif self.username:
            self.id = self.username.value
        self.nodeLabel = self.display_name or self.id
        return self

This pattern keeps the public constructor simple while guaranteeing consistent internal representation.

Runtime Discovery and the Type Registry

The framework consumes types dynamically through the registry rather than hard-coding imports.

Startup Auto-Discovery

flowsint_types.__init__ invokes load_all_types() to import every module and fire all decorators.

from flowsint_types.registry import load_all_types

load_all_types()   # imports every module; all @flowsint_type decorators fire

Dynamic Type Lookup

User code or graph layers can retrieve a class by its lowercase or exact-case name.

from flowsint_types.registry import get_type

Cls = get_type("domain")            # case-insensitive lookup

domain_obj = Cls(domain="sub.example.org")
print(domain_obj.nodeLabel)         # → "sub.example.org"

Creating Instances in Practice

Because validators run automatically, callers pass primitive data and receive fully normalized entities.

from flowsint_types.domain import Domain

raw = "example.com"
if Domain.detect(raw):
    d = Domain.from_string(raw)          # triggers field & model validators

    print(d.nodeLabel)                  # → "example.com"

    print(d.root)                       # → True (root domain)

Complex types like SocialAccount accept mixed primitive and object inputs.

from flowsint_types.social_account import SocialAccount
from flowsint_types.username import Username

acct = SocialAccount(
    username="alice",                     # automatically converted to Username

    platform="twitter",
    display_name="Alice Smith"
)
print(acct.id)          # → "alice@twitter"

print(acct.nodeLabel)   # → "Alice Smith (@alice)"

Summary

• FlowsintType in flowsint_base.py provides the universal base class with extra="allow" and the nodeLabel field.
• The @flowsint_type decorator and TYPE_REGISTRY in registry.py enable automatic, case-sensitive and lowercase registration of every entity class.
• Field validators sanitize raw input before instantiation, while model validators compute derived fields such as composite IDs and graph labels.
• The startup load_all_types() routine removes the need for manual imports, letting the core platform and enrichers discover types dynamically.
• Concrete modules like domain.py and social_account.py demonstrate production-ready patterns for hostname normalization, root-domain detection, username coercion, and deterministic ID generation.

Frequently Asked Questions

What is FlowsintType and why does it matter?

FlowsintType is the Pydantic base class defined in flowsint_base.py from which every investigative entity inherits. It matters because it standardizes graph-oriented metadata—specifically the nodeLabel field—and configures extra="allow" so unexpected keys do not fail validation.

How does Flowsint register types automatically?

Flowsint registers types through the @flowsint_type decorator implemented in registry.py. When load_all_types() runs at startup, it imports all modules in the flowsint_types package, which triggers the decorator and populates TYPE_REGISTRY with both exact-case and lowercase name mappings.

Can I look up a Flowsint type by a string name at runtime?

Yes. The get_type function in registry.py performs case-insensitive lookups against TYPE_REGISTRY. This lets graph queries and enrichers instantiate the correct class without hard-coding imports.

What is the difference between @field_validator and @model_validator in Flowsint?

@field_validator runs before the model is built and is used for input sanitization, such as converting a string to a Username object. @model_validator(mode="after") runs after instantiation and is used for post-processing, such as computing nodeLabel or composing a primary ID from multiple fields.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}