feat(api): custom types updates

flowsint-api/alembic/versions/bac5764d4496_add_icon_and_color_to_custom_types.py

@@ -0,0 +1,42 @@
+"""add icon and color to custom types
+
+Revision ID: bac5764d4496
+Revises: f5fae279ec04
+Create Date: 2026-02-14 12:10:26.415916
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision: str = 'bac5764d4496'
+down_revision: Union[str, None] = 'f5fae279ec04'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('custom_types', sa.Column('icon', sa.String(), nullable=True))
+    op.add_column('custom_types', sa.Column('color', sa.String(), nullable=True))
+    op.alter_column('enricher_templates', 'content',
+               existing_type=postgresql.JSONB(astext_type=sa.Text()),
+               type_=sa.JSON(),
+               existing_nullable=False)
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.alter_column('enricher_templates', 'content',
+               existing_type=sa.JSON(),
+               type_=postgresql.JSONB(astext_type=sa.Text()),
+               existing_nullable=False)
+    op.drop_column('custom_types', 'color')
+    op.drop_column('custom_types', 'icon')
+    # ### end Alembic commands ###

flowsint-api/app/api/routes/custom_types.py

@@ -1,32 +1,33 @@
 """API routes for custom types management."""
-from uuid import UUID
-from typing import List
-from fastapi import APIRouter, HTTPException, Depends, status
-from sqlalchemy.orm import Session
 
-from flowsint_core.core.postgre_db import get_db
+from typing import List
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, HTTPException, status
 from flowsint_core.core.models import Profile
+from flowsint_core.core.postgre_db import get_db
 from flowsint_core.core.services import (
-    create_custom_type_service,
+    ConflictError,
     NotFoundError,
     ValidationError,
-    ConflictError,
+    create_custom_type_service,
 )
+from sqlalchemy.orm import Session
+
 from app.api.deps import get_current_user
 from app.api.schemas.custom_type import (
     CustomTypeCreate,
-    CustomTypeUpdate,
     CustomTypeRead,
+    CustomTypeUpdate,
     CustomTypeValidatePayload,
     CustomTypeValidateResponse,
 )
 from app.utils.custom_types import (
+    calculate_schema_checksum,
     validate_json_schema,
     validate_payload_against_schema,
-    calculate_schema_checksum,
 )
 
-
 router = APIRouter()
 
 
@@ -110,6 +111,8 @@ def update_custom_type(
             custom_type_id=id,
             user_id=current_user.id,
             name=update_data.name,
+            icon=update_data.icon,
+            color=update_data.color,
             json_schema=update_data.json_schema,
             description=update_data.description,
             status=update_data.status,

flowsint-api/app/api/routes/enrichers.py

@@ -9,6 +9,7 @@ from flowsint_core.core.services import (
     create_enricher_service,
     create_enricher_template_service,
 )
+from flowsint_core.core.services.type_registry_service import create_type_registry_service
 from flowsint_enrichers import ENRICHER_REGISTRY, load_all_enrichers
 from pydantic import BaseModel
 from sqlalchemy.orm import Session
@@ -48,7 +49,9 @@ async def launch_enricher(
 ):
     try:
         # Retrieve nodes from Neo4J by their element IDs
-        graph_service = create_graph_service(sketch_id=payload.sketch_id)
+        type_registry = create_type_registry_service(db)
+        resolver = type_registry.build_type_resolver(current_user.id)
+        graph_service = create_graph_service(sketch_id=payload.sketch_id, type_resolver=resolver)
         entities = graph_service.get_nodes_by_ids_for_task(payload.node_ids)
 
         # Send deserialized nodes

flowsint-api/app/api/routes/flows.py

@@ -11,6 +11,9 @@ from flowsint_core.core.services import (
     PermissionDeniedError,
     create_flow_service,
 )
+from flowsint_core.core.services.type_registry_service import (
+    create_type_registry_service,
+)
 from flowsint_core.core.types import FlowBranch, FlowEdge, FlowNode, FlowStep
 from flowsint_core.utils import extract_input_schema_flow
 from flowsint_enrichers import ENRICHER_REGISTRY, load_all_enrichers
@@ -202,7 +205,11 @@ async def launch_flow(
         service.get_sketch_for_launch(payload.sketch_id, current_user.id)
 
         # Retrieve entities from Neo4J by their element IDs
-        graph_service = create_graph_service(sketch_id=payload.sketch_id)
+        type_registry = create_type_registry_service(db)
+        resolver = type_registry.build_type_resolver(current_user.id)
+        graph_service = create_graph_service(
+            sketch_id=payload.sketch_id, type_resolver=resolver
+        )
         entities = graph_service.get_nodes_by_ids_for_task(payload.node_ids)
 
         # Compute flow branches
@@ -289,6 +296,7 @@ def compute_flow_branches(
                     FlowStep(
                         nodeId="error",
                         inputs={},
+                        params={},
                         type="error",
                         outputs={},
                         status="error",

flowsint-api/app/api/routes/sketches.py

@@ -21,6 +21,7 @@ from flowsint_core.core.services import (
     ValidationError,
     DatabaseError,
 )
+from flowsint_core.core.services.type_registry_service import create_type_registry_service
 from flowsint_core.imports import (
     EntityMapping,
     ImportService,
@@ -431,7 +432,11 @@ async def analyze_import_file(
         raise HTTPException(status_code=400, detail=f"Failed to read file: {str(e)}")
 
     try:
-        result = ImportService.analyze_file(
+        type_registry = create_type_registry_service(db)
+        resolver = type_registry.build_type_resolver(current_user.id)
+        graph_service = create_graph_service(sketch_id=sketch_id, enable_batching=False, type_resolver=resolver)
+        import_service = create_import_service(graph_service)
+        result = import_service.analyze_file(
             file_content=content,
             filename=file.filename or "unknown.txt",
         )
@@ -487,7 +492,9 @@ async def execute_import(
         for m in entity_mapping_inputs
     ]
 
-    graph_service = create_graph_service(sketch_id=sketch_id, enable_batching=False)
+    type_registry = create_type_registry_service(db)
+    resolver = type_registry.build_type_resolver(current_user.id)
+    graph_service = create_graph_service(sketch_id=sketch_id, enable_batching=False, type_resolver=resolver)
     import_service = create_import_service(graph_service)
 
     try:

flowsint-api/app/api/schemas/custom_type.py

@@ -1,20 +1,31 @@
-from typing import Optional, Dict, Any
-from pydantic import BaseModel, UUID4, Field, field_validator
 from datetime import datetime
+from typing import Any, Dict, Optional
+
+from pydantic import UUID4, BaseModel, Field, field_validator
+
 from .base import ORMBase
 
 
 class CustomTypeCreate(BaseModel):
     """Schema for creating a new custom type."""
-    name: str = Field(..., min_length=1, max_length=255, description="Name of the custom type")
-    json_schema: Dict[str, Any] = Field(..., description="JSON Schema definition", alias="schema")
-    description: Optional[str] = Field(None, description="Optional description of the custom type")
-    status: Optional[str] = Field("draft", description="Status: draft or published")
+
+    name: str = Field(
+        ..., min_length=1, max_length=255, description="Name of the custom type"
+    )
+    json_schema: Dict[str, Any] = Field(
+        ..., description="JSON Schema definition", alias="schema"
+    )
+    description: Optional[str] = Field(
+        None, description="Optional description of the custom type"
+    )
+    status: str = Field("draft", description="Status of the custom type")
+    color: str = Field("#8E9E8C", description="Default color")
+    icon: str = Field("Minus", description="Default icon")
 
     class Config:
         populate_by_name = True
 
-    @field_validator('status')
+    @field_validator("status")
     @classmethod
     def validate_status(cls, v: str) -> str:
         if v not in ["draft", "published", "archived"]:
@@ -24,15 +35,18 @@ class CustomTypeCreate(BaseModel):
 
 class CustomTypeUpdate(BaseModel):
     """Schema for updating an existing custom type."""
+
     name: Optional[str] = Field(None, min_length=1, max_length=255)
     json_schema: Optional[Dict[str, Any]] = Field(None, alias="schema")
     description: Optional[str] = None
     status: Optional[str] = None
+    color: Optional[str] = None
+    icon: Optional[str] = None
 
     class Config:
         populate_by_name = True
 
-    @field_validator('status')
+    @field_validator("status")
     @classmethod
     def validate_status(cls, v: Optional[str]) -> Optional[str]:
         if v is not None and v not in ["draft", "published", "archived"]:
@@ -42,9 +56,12 @@ class CustomTypeUpdate(BaseModel):
 
 class CustomTypeRead(ORMBase):
     """Schema for reading a custom type."""
+
     id: UUID4
     name: str
     owner_id: UUID4
+    color: Optional[str]
+    icon: Optional[str]
     json_schema: Dict[str, Any] = Field(..., alias="schema")
     status: str
     checksum: Optional[str]
@@ -58,10 +75,14 @@ class CustomTypeRead(ORMBase):
 
 class CustomTypeValidatePayload(BaseModel):
     """Schema for validating a payload against a custom type schema."""
-    payload: Dict[str, Any] = Field(..., description="Data to validate against the schema")
+
+    payload: Dict[str, Any] = Field(
+        ..., description="Data to validate against the schema"
+    )
 
 
 class CustomTypeValidateResponse(BaseModel):
     """Response schema for validation."""
+
     valid: bool
     errors: Optional[list[str]] = None