from __future__ import annotations from enum import Enum from typing import Any from pydantic import BaseModel, ConfigDict, Field, model_validator class StrictModel(BaseModel): model_config = ConfigDict(extra="forbid", populate_by_name=True) class EntityKind(str, Enum): master = "master" transaction = "transaction" activity = "activity" bridge = "bridge" class FieldType(str, Enum): text = "text" long_text = "long_text" number = "number" amount = "amount" date = "date" datetime = "datetime" member = "member" department = "department" single_select = "single_select" multi_select = "multi_select" phone = "phone" email = "email" address = "address" attachment = "attachment" boolean = "boolean" q_linker = "q_linker" code_block = "code_block" relation = "relation" subtable = "subtable" class WorkflowNodeType(str, Enum): start = "start" branch = "branch" audit = "audit" fill = "fill" copy = "copy" webhook = "webhook" qsource = "qsource" condition = "condition" class ViewType(str, Enum): table = "table" card = "card" board = "board" gantt = "gantt" hierarchy = "hierarchy" class ChartType(str, Enum): bar = "bar" line = "line" pie = "pie" summary = "summary" data = "data" target = "target" funnel = "funnel" class RequirementPriority(str, Enum): must = "must" should = "should" could = "could" class MetricAggregate(str, Enum): count = "count" sum = "sum" avg = "avg" distinct_count = "distinct_count" class PortalSourceType(str, Enum): chart = "chart" grid = "grid" view = "view" filter = "filter" text = "text" link = "link" class NavigationTargetType(str, Enum): package = "package" app = "app" chart = "chart" portal = "portal" view = "view" custom_url = "custom_url" class PackageSpec(StrictModel): enabled: bool = True name: str | None = None icon: str | None = None color: str | None = None ordinal: int = 1 class BusinessContextSpec(StrictModel): industry: str | None = None subdomain: str | None = None scenario: str | None = None target_users: list[str] = Field(default_factory=list) core_goals: list[str] = Field(default_factory=list) terminology: dict[str, str] = Field(default_factory=dict) non_goals: list[str] = Field(default_factory=list) class FieldSpec(StrictModel): field_id: str label: str type: FieldType required: bool = False unique: bool = False description: str | None = None options: list[str] = Field(default_factory=list) target_entity_id: str | None = None target_field_id: str | None = None subfields: list["FieldSpec"] = Field(default_factory=list) config: dict[str, Any] = Field(default_factory=dict) @model_validator(mode="after") def validate_field_shape(self) -> "FieldSpec": if self.type == FieldType.relation and not self.target_entity_id: raise ValueError("relation field must declare target_entity_id") if self.type != FieldType.relation and self.target_entity_id: raise ValueError("target_entity_id is only allowed for relation fields") if self.type == FieldType.subtable and not self.subfields: raise ValueError("subtable field must declare subfields") if self.type != FieldType.subtable and self.subfields: raise ValueError("subfields are only allowed for subtable fields") return self class FormLayoutRowSpec(StrictModel): field_ids: list[str] = Field(default_factory=list) config: dict[str, Any] = Field(default_factory=dict) @model_validator(mode="after") def validate_row(self) -> "FormLayoutRowSpec": if not self.field_ids: raise ValueError("form layout row must include at least one field_id") return self class FormLayoutSectionSpec(StrictModel): section_id: str title: str rows: list[FormLayoutRowSpec] = Field(default_factory=list) config: dict[str, Any] = Field(default_factory=dict) @model_validator(mode="after") def validate_section(self) -> "FormLayoutSectionSpec": if not self.rows: raise ValueError("form layout section must include at least one row") return self class FormLayoutSpec(StrictModel): rows: list[FormLayoutRowSpec] = Field(default_factory=list) sections: list[FormLayoutSectionSpec] = Field(default_factory=list) @model_validator(mode="after") def validate_layout(self) -> "FormLayoutSpec": if not self.rows and not self.sections: raise ValueError("form_layout must include rows or sections") return self class RelationSpec(StrictModel): relation_id: str field_id: str target_entity_id: str target_field_id: str | None = None relation_type: str = "many_to_one" config: dict[str, Any] = Field(default_factory=dict) class LifecycleStageSpec(StrictModel): stage_id: str name: str terminal: bool = False color: str | None = None probability: float | None = None class WorkflowNodeSpec(StrictModel): node_id: str name: str node_type: WorkflowNodeType parent_node_id: str | None = None branch_parent_id: str | None = None branch_index: int | None = None assignees: dict[str, Any] = Field(default_factory=dict) config: dict[str, Any] = Field(default_factory=dict) @model_validator(mode="after") def validate_branch_shape(self) -> "WorkflowNodeSpec": if self.branch_index is not None and self.branch_index <= 0: raise ValueError("branch_index must be positive") if self.branch_parent_id and self.branch_index is None: raise ValueError("branch_index is required when branch_parent_id is set") if self.branch_index is not None and not self.branch_parent_id: raise ValueError("branch_parent_id is required when branch_index is set") if self.node_type == WorkflowNodeType.branch and self.branch_parent_id: raise ValueError("branch node cannot declare branch_parent_id") if self.node_type == WorkflowNodeType.branch and self.branch_index is not None: raise ValueError("branch node cannot declare branch_index") return self class WorkflowSpec(StrictModel): enabled: bool = True global_settings: dict[str, Any] = Field(default_factory=dict) nodes: list[WorkflowNodeSpec] = Field(default_factory=list) class RoleEntityScopeSpec(StrictModel): entity_id: str can_create: bool = True can_edit: bool = True can_delete: bool = False can_approve: bool = False visible_field_ids: list[str] = Field(default_factory=list) config: dict[str, Any] = Field(default_factory=dict) class RoleSpec(StrictModel): role_id: str name: str description: str | None = None icon: str | None = None users: list[int] = Field(default_factory=list) entity_scopes: list[RoleEntityScopeSpec] = Field(default_factory=list) config: dict[str, Any] = Field(default_factory=dict) class RequirementSpec(StrictModel): requirement_id: str title: str summary: str | None = None priority: RequirementPriority = RequirementPriority.must entity_ids: list[str] = Field(default_factory=list) acceptance_criteria: list[str] = Field(default_factory=list) tags: list[str] = Field(default_factory=list) config: dict[str, Any] = Field(default_factory=dict) class ViewSpec(StrictModel): view_id: str name: str type: ViewType field_ids: list[str] = Field(default_factory=list) group_by_field_id: str | None = None sort: list[dict[str, Any]] = Field(default_factory=list) filters: list[dict[str, Any]] = Field(default_factory=list) column_widths: dict[str, int] = Field(default_factory=dict) member_config: dict[str, Any] = Field(default_factory=dict) apply_config: dict[str, Any] = Field(default_factory=dict) config: dict[str, Any] = Field(default_factory=dict) being_default: bool = False class ChartSpec(StrictModel): chart_id: str name: str chart_type: ChartType dimension_field_ids: list[str] = Field(default_factory=list) indicator_field_ids: list[str] = Field(default_factory=list) filters: list[dict[str, Any]] = Field(default_factory=list) question_config: list[dict[str, Any]] = Field(default_factory=list) user_config: list[dict[str, Any]] = Field(default_factory=list) config: dict[str, Any] = Field(default_factory=dict) class SampleRecordSpec(StrictModel): record_id: str | None = None values: dict[str, Any] = Field(default_factory=dict) submit_type: int = 1 @model_validator(mode="after") def validate_values(self) -> "SampleRecordSpec": if not self.values: raise ValueError("sample record must include values") if self.submit_type not in (0, 1): raise ValueError("sample record submit_type must be 0 or 1") return self class MetricSpec(StrictModel): metric_id: str name: str entity_id: str aggregate: MetricAggregate = MetricAggregate.count field_id: str | None = None group_by_field_id: str | None = None time_field_id: str | None = None visualization_preference: ChartType | None = None target_value: float | None = None description: str | None = None config: dict[str, Any] = Field(default_factory=dict) class PortalSectionSpec(StrictModel): section_id: str title: str source_type: PortalSourceType entity_id: str | None = None chart_id: str | None = None view_id: str | None = None text: str | None = None url: str | None = None config: dict[str, Any] = Field(default_factory=dict) class PortalSpec(StrictModel): enabled: bool = True name: str | None = None icon: str | None = None color: str | None = None sections: list[PortalSectionSpec] = Field(default_factory=list) config: dict[str, Any] = Field(default_factory=dict) class NavigationItemSpec(StrictModel): item_id: str title: str target_type: NavigationTargetType entity_id: str | None = None chart_id: str | None = None view_id: str | None = None url: str | None = None icon: str | None = None visible: bool = True children: list["NavigationItemSpec"] = Field(default_factory=list) config: dict[str, Any] = Field(default_factory=dict) class NavigationSpec(StrictModel): enabled: bool = True items: list[NavigationItemSpec] = Field(default_factory=list) class EntitySpec(StrictModel): entity_id: str display_name: str plural_name: str | None = None kind: EntityKind aliases: list[str] = Field(default_factory=list) business_purpose: str | None = None description: str | None = None fields: list[FieldSpec] = Field(default_factory=list) form_layout: FormLayoutSpec | None = None relations: list[RelationSpec] = Field(default_factory=list) workflow: WorkflowSpec | None = None views: list[ViewSpec] = Field(default_factory=list) charts: list[ChartSpec] = Field(default_factory=list) sample_records: list[SampleRecordSpec] = Field(default_factory=list) title_field_id: str | None = None status_field_id: str | None = None owner_field_id: str | None = None start_field_id: str | None = None end_field_id: str | None = None lifecycle_stages: list[LifecycleStageSpec] = Field(default_factory=list) acceptance_criteria: list[str] = Field(default_factory=list) portal_exposure: bool = True navigation_exposure: bool = True icon: str | None = None color: str | None = None ordinal: int | None = None metadata: dict[str, Any] = Field(default_factory=dict) @model_validator(mode="after") def validate_entity_uniqueness(self) -> "EntitySpec": field_ids = [field.field_id for field in self.fields] if len(field_ids) != len(set(field_ids)): raise ValueError(f"entity '{self.entity_id}' has duplicate field_id values") field_labels = [field.label for field in self.fields] if len(field_labels) != len(set(field_labels)): raise ValueError(f"entity '{self.entity_id}' has duplicate field labels") relation_ids = [relation.relation_id for relation in self.relations] if len(relation_ids) != len(set(relation_ids)): raise ValueError(f"entity '{self.entity_id}' has duplicate relation_id values") view_ids = [view.view_id for view in self.views] if len(view_ids) != len(set(view_ids)): raise ValueError(f"entity '{self.entity_id}' has duplicate view_id values") chart_ids = [chart.chart_id for chart in self.charts] if len(chart_ids) != len(set(chart_ids)): raise ValueError(f"entity '{self.entity_id}' has duplicate chart_id values") node_ids = [node.node_id for node in self.workflow.nodes] if self.workflow else [] if len(node_ids) != len(set(node_ids)): raise ValueError(f"entity '{self.entity_id}' has duplicate workflow node_id values") stage_ids = [stage.stage_id for stage in self.lifecycle_stages] if len(stage_ids) != len(set(stage_ids)): raise ValueError(f"entity '{self.entity_id}' has duplicate lifecycle stage_id values") field_map = {field.field_id: field for field in self.fields} for index, sample_record in enumerate(self.sample_records, start=1): record_name = sample_record.record_id or f"sample#{index}" for field_id, raw_value in sample_record.values.items(): field = field_map.get(field_id) if field is None: raise ValueError(f"entity '{self.entity_id}' sample record '{record_name}' references unknown field '{field_id}'") _validate_sample_value(self.entity_id, record_name, field, raw_value) return self class PublishPolicy(StrictModel): apps: bool = True portal: bool = True navigation: bool = True class Preferences(StrictModel): multi_app: bool | None = None create_package: bool = True create_portal: bool = True create_navigation: bool = True naming_style: str = "title" class SolutionSpec(StrictModel): solution_name: str summary: str | None = None business_context: BusinessContextSpec = Field(default_factory=BusinessContextSpec) package: PackageSpec = Field(default_factory=PackageSpec) entities: list[EntitySpec] roles: list[RoleSpec] = Field(default_factory=list) requirements: list[RequirementSpec] = Field(default_factory=list) success_metrics: list[MetricSpec] = Field(default_factory=list) portal: PortalSpec = Field(default_factory=PortalSpec) navigation: NavigationSpec = Field(default_factory=NavigationSpec) publish_policy: PublishPolicy = Field(default_factory=PublishPolicy) preferences: Preferences = Field(default_factory=Preferences) assumptions: list[str] = Field(default_factory=list) constraints: list[str] = Field(default_factory=list) metadata: dict[str, Any] = Field(default_factory=dict) @model_validator(mode="after") def validate_cross_references(self) -> "SolutionSpec": entity_map = {entity.entity_id: entity for entity in self.entities} role_ids = {role.role_id for role in self.roles} if len(entity_map) != len(self.entities): raise ValueError("duplicate entity_id values are not allowed") for entity in self.entities: field_ids = self._effective_field_ids(entity) title_field_id = entity.title_field_id or "title" status_field_id = entity.status_field_id or "status" if entity.relations: raise ValueError( f"entity '{entity.entity_id}' uses deprecated data relations; use relation fields (reference fields) instead" ) if entity.kind in (EntityKind.master, EntityKind.transaction) and title_field_id not in field_ids: raise ValueError( f"entity '{entity.entity_id}' must declare title field '{title_field_id}' explicitly for AI-authored DSL" ) if ((entity.workflow and entity.workflow.enabled) or entity.lifecycle_stages) and status_field_id not in field_ids: raise ValueError( f"entity '{entity.entity_id}' must declare status field '{status_field_id}' explicitly for AI-authored DSL" ) for field in entity.fields: if field.type == FieldType.relation and field.target_entity_id not in entity_map: raise ValueError(f"entity '{entity.entity_id}' relation field '{field.field_id}' targets unknown entity '{field.target_entity_id}'") if field.type == FieldType.relation and field.target_entity_id in entity_map: target_entity = entity_map[field.target_entity_id] target_field_ids = self._effective_field_ids(target_entity) target_field_id = field.target_field_id or target_entity.title_field_id or "title" if target_field_id not in target_field_ids: raise ValueError( f"entity '{entity.entity_id}' relation field '{field.field_id}' targets unknown field '{target_field_id}' on entity '{field.target_entity_id}'" ) if field.type == FieldType.subtable: sub_ids = [subfield.field_id for subfield in field.subfields] if len(sub_ids) != len(set(sub_ids)): raise ValueError(f"entity '{entity.entity_id}' subtable field '{field.field_id}' has duplicate subfield ids") self._validate_special_field_ref(entity.entity_id, "title_field_id", entity.title_field_id, field_ids) self._validate_special_field_ref(entity.entity_id, "status_field_id", entity.status_field_id, field_ids) self._validate_special_field_ref(entity.entity_id, "owner_field_id", entity.owner_field_id, field_ids) self._validate_special_field_ref(entity.entity_id, "start_field_id", entity.start_field_id, field_ids) self._validate_special_field_ref(entity.entity_id, "end_field_id", entity.end_field_id, field_ids) for relation in entity.relations: if relation.field_id not in field_ids: raise ValueError(f"entity '{entity.entity_id}' relation '{relation.relation_id}' references unknown field '{relation.field_id}'") if relation.target_entity_id not in entity_map: raise ValueError(f"entity '{entity.entity_id}' relation '{relation.relation_id}' targets unknown entity '{relation.target_entity_id}'") for view in entity.views: self._validate_field_refs(entity.entity_id, view.view_id, view.field_ids, field_ids, "view") if view.group_by_field_id and view.group_by_field_id not in field_ids: raise ValueError(f"entity '{entity.entity_id}' view '{view.view_id}' references unknown field '{view.group_by_field_id}'") for field_id in view.column_widths: if field_id not in field_ids: raise ValueError(f"entity '{entity.entity_id}' view '{view.view_id}' references unknown field '{field_id}'") if entity.form_layout: layout_field_ids: list[str] = [] for row in entity.form_layout.rows: self._validate_field_refs(entity.entity_id, "form_layout", row.field_ids, field_ids, "form_layout") layout_field_ids.extend(row.field_ids) for section in entity.form_layout.sections: for row in section.rows: self._validate_field_refs(entity.entity_id, f"form_layout.{section.section_id}", row.field_ids, field_ids, "form_layout") layout_field_ids.extend(row.field_ids) duplicated_layout_fields = {field_id for field_id in layout_field_ids if layout_field_ids.count(field_id) > 1} if duplicated_layout_fields: duplicates = ", ".join(sorted(duplicated_layout_fields)) raise ValueError(f"entity '{entity.entity_id}' form_layout references duplicated fields: {duplicates}") for chart in entity.charts: self._validate_field_refs(entity.entity_id, chart.chart_id, chart.dimension_field_ids, field_ids, "chart") self._validate_field_refs(entity.entity_id, chart.chart_id, chart.indicator_field_ids, field_ids, "chart") if entity.workflow: node_map = {node.node_id: node for node in entity.workflow.nodes} node_ids = set(node_map) for node in entity.workflow.nodes: if node.parent_node_id and node.parent_node_id not in node_ids: raise ValueError( f"entity '{entity.entity_id}' workflow node '{node.node_id}' references unknown parent '{node.parent_node_id}'" ) if node.branch_parent_id and node.branch_parent_id not in node_ids: raise ValueError( f"entity '{entity.entity_id}' workflow node '{node.node_id}' references unknown branch parent '{node.branch_parent_id}'" ) if node.branch_parent_id: branch_parent = node_map[node.branch_parent_id] if branch_parent.node_type != WorkflowNodeType.branch: raise ValueError( f"entity '{entity.entity_id}' workflow node '{node.node_id}' must reference a branch node in branch_parent_id" ) for role_ref in node.assignees.get("role_refs", []): if role_ref not in role_ids: raise ValueError( f"entity '{entity.entity_id}' workflow node '{node.node_id}' references unknown role '{role_ref}'" ) for role in self.roles: for scope in role.entity_scopes: if scope.entity_id not in entity_map: raise ValueError(f"role '{role.role_id}' references unknown entity '{scope.entity_id}'") valid_field_ids = self._effective_field_ids(entity_map[scope.entity_id]) self._validate_field_refs(scope.entity_id, role.role_id, scope.visible_field_ids, valid_field_ids, "role") for requirement in self.requirements: for entity_id in requirement.entity_ids: if entity_id not in entity_map: raise ValueError(f"requirement '{requirement.requirement_id}' references unknown entity '{entity_id}'") for metric in self.success_metrics: if metric.entity_id not in entity_map: raise ValueError(f"metric '{metric.metric_id}' references unknown entity '{metric.entity_id}'") valid_field_ids = self._effective_field_ids(entity_map[metric.entity_id]) self._validate_special_field_ref(metric.entity_id, "field_id", metric.field_id, valid_field_ids) self._validate_special_field_ref(metric.entity_id, "group_by_field_id", metric.group_by_field_id, valid_field_ids) self._validate_special_field_ref(metric.entity_id, "time_field_id", metric.time_field_id, valid_field_ids) for section in self.portal.sections: self._validate_portal_ref(section, entity_map) for item in self.navigation.items: self._validate_navigation_ref(item, entity_map) return self @staticmethod def _validate_field_refs(entity_id: str, resource_id: str, refs: list[str], valid_field_ids: set[str], resource_type: str) -> None: for field_id in refs: if field_id not in valid_field_ids: raise ValueError(f"entity '{entity_id}' {resource_type} '{resource_id}' references unknown field '{field_id}'") @staticmethod def _validate_special_field_ref(entity_id: str, field_name: str, ref: str | None, valid_field_ids: set[str]) -> None: if ref is not None and ref not in valid_field_ids: raise ValueError(f"entity '{entity_id}' references unknown field '{ref}' in '{field_name}'") @staticmethod def _effective_field_ids(entity: EntitySpec) -> set[str]: return {field.field_id for field in entity.fields} @classmethod def _validate_portal_ref(cls, section: PortalSectionSpec, entity_map: dict[str, EntitySpec]) -> None: if section.entity_id and section.entity_id not in entity_map: raise ValueError(f"portal section '{section.section_id}' references unknown entity '{section.entity_id}'") if section.chart_id and section.entity_id: chart_ids = {chart.chart_id for chart in entity_map[section.entity_id].charts} if section.chart_id not in chart_ids: raise ValueError(f"portal section '{section.section_id}' references unknown chart '{section.chart_id}'") if section.view_id and section.entity_id: view_ids = {view.view_id for view in entity_map[section.entity_id].views} if section.view_id not in view_ids: raise ValueError(f"portal section '{section.section_id}' references unknown view '{section.view_id}'") @classmethod def _validate_navigation_ref(cls, item: NavigationItemSpec, entity_map: dict[str, EntitySpec]) -> None: if item.entity_id and item.entity_id not in entity_map: raise ValueError(f"navigation item '{item.item_id}' references unknown entity '{item.entity_id}'") if item.chart_id and item.entity_id: chart_ids = {chart.chart_id for chart in entity_map[item.entity_id].charts} if item.chart_id not in chart_ids: raise ValueError(f"navigation item '{item.item_id}' references unknown chart '{item.chart_id}'") if item.view_id and item.entity_id: view_ids = {view.view_id for view in entity_map[item.entity_id].views} if item.view_id not in view_ids: raise ValueError(f"navigation item '{item.item_id}' references unknown view '{item.view_id}'") for child in item.children: cls._validate_navigation_ref(child, entity_map) class AppFlowBuildSpec(StrictModel): solution_name: str summary: str | None = None business_context: BusinessContextSpec = Field(default_factory=BusinessContextSpec) package: PackageSpec = Field(default_factory=PackageSpec) entities: list[EntitySpec] roles: list[RoleSpec] = Field(default_factory=list) publish_policy: PublishPolicy = Field(default_factory=PublishPolicy) preferences: Preferences = Field(default_factory=Preferences) assumptions: list[str] = Field(default_factory=list) constraints: list[str] = Field(default_factory=list) metadata: dict[str, Any] = Field(default_factory=dict) @model_validator(mode="after") def validate_entities(self) -> "AppFlowBuildSpec": if not self.entities: raise ValueError("entities must be a non-empty list") return self class AppEntityBuildSpec(StrictModel): entity_id: str display_name: str plural_name: str | None = None kind: EntityKind aliases: list[str] = Field(default_factory=list) business_purpose: str | None = None description: str | None = None fields: list[FieldSpec] = Field(default_factory=list) form_layout: FormLayoutSpec | None = None sample_records: list[SampleRecordSpec] = Field(default_factory=list) title_field_id: str | None = None status_field_id: str | None = None owner_field_id: str | None = None start_field_id: str | None = None end_field_id: str | None = None acceptance_criteria: list[str] = Field(default_factory=list) portal_exposure: bool = True navigation_exposure: bool = True icon: str | None = None color: str | None = None ordinal: int | None = None metadata: dict[str, Any] = Field(default_factory=dict) @model_validator(mode="after") def validate_entity_shape(self) -> "AppEntityBuildSpec": if not self.fields: raise ValueError(f"entity '{self.entity_id}' must declare fields") field_ids = [field.field_id for field in self.fields] if len(field_ids) != len(set(field_ids)): raise ValueError(f"entity '{self.entity_id}' has duplicate field_id values") field_labels = [field.label for field in self.fields] if len(field_labels) != len(set(field_labels)): raise ValueError(f"entity '{self.entity_id}' has duplicate field labels") field_map = {field.field_id: field for field in self.fields} for index, sample_record in enumerate(self.sample_records, start=1): record_name = sample_record.record_id or f"sample#{index}" for field_id, raw_value in sample_record.values.items(): field = field_map.get(field_id) if field is None: raise ValueError(f"entity '{self.entity_id}' sample record '{record_name}' references unknown field '{field_id}'") _validate_sample_value(self.entity_id, record_name, field, raw_value) return self class AppBuildSpec(StrictModel): solution_name: str summary: str | None = None business_context: BusinessContextSpec = Field(default_factory=BusinessContextSpec) package: PackageSpec = Field(default_factory=PackageSpec) entities: list[AppEntityBuildSpec] roles: list[RoleSpec] = Field(default_factory=list) publish_policy: PublishPolicy = Field(default_factory=PublishPolicy) preferences: Preferences = Field(default_factory=Preferences) assumptions: list[str] = Field(default_factory=list) constraints: list[str] = Field(default_factory=list) metadata: dict[str, Any] = Field(default_factory=dict) @model_validator(mode="after") def validate_entities(self) -> "AppBuildSpec": if not self.entities: raise ValueError("entities must be a non-empty list") return self class FlowEntityBuildSpec(StrictModel): entity_id: str workflow: WorkflowSpec | None = None lifecycle_stages: list[LifecycleStageSpec] = Field(default_factory=list) status_field_id: str | None = None owner_field_id: str | None = None start_field_id: str | None = None end_field_id: str | None = None metadata: dict[str, Any] = Field(default_factory=dict) @model_validator(mode="after") def validate_flow_shape(self) -> "FlowEntityBuildSpec": if self.workflow is None and not self.lifecycle_stages: raise ValueError(f"entity '{self.entity_id}' must declare workflow or lifecycle_stages") if self.workflow and self.workflow.enabled and not self.workflow.nodes and not self.lifecycle_stages: raise ValueError(f"entity '{self.entity_id}' workflow.nodes must be a non-empty list when workflow.enabled is true") return self class FlowBuildSpec(StrictModel): solution_name: str | None = None entities: list[FlowEntityBuildSpec] roles: list[RoleSpec] = Field(default_factory=list) publish_policy: PublishPolicy = Field(default_factory=PublishPolicy) metadata: dict[str, Any] = Field(default_factory=dict) @model_validator(mode="after") def validate_entities(self) -> "FlowBuildSpec": if not self.entities: raise ValueError("entities must be a non-empty list") return self class EntityViewSpec(StrictModel): entity_id: str views: list[ViewSpec] = Field(default_factory=list) @model_validator(mode="after") def validate_views(self) -> "EntityViewSpec": if not self.views: raise ValueError("views must be a non-empty list") return self class ViewsBuildSpec(StrictModel): solution_name: str | None = None entities: list[EntityViewSpec] metadata: dict[str, Any] = Field(default_factory=dict) @model_validator(mode="after") def validate_entities(self) -> "ViewsBuildSpec": if not self.entities: raise ValueError("entities must be a non-empty list") return self class EntityAnalyticsSpec(StrictModel): entity_id: str charts: list[ChartSpec] = Field(default_factory=list) @model_validator(mode="after") def validate_charts(self) -> "EntityAnalyticsSpec": if not self.charts: raise ValueError("charts must be a non-empty list") return self class AnalyticsPortalBuildSpec(StrictModel): solution_name: str | None = None entities: list[EntityAnalyticsSpec] = Field(default_factory=list) portal: PortalSpec = Field(default_factory=PortalSpec) publish_policy: PublishPolicy = Field(default_factory=PublishPolicy) metadata: dict[str, Any] = Field(default_factory=dict) @model_validator(mode="after") def validate_payload(self) -> "AnalyticsPortalBuildSpec": if not self.entities and not self.portal.sections: raise ValueError("analytics build must declare charts or portal sections") return self class NavigationBuildSpec(StrictModel): solution_name: str | None = None navigation: NavigationSpec = Field(default_factory=NavigationSpec) publish_policy: PublishPolicy = Field(default_factory=PublishPolicy) metadata: dict[str, Any] = Field(default_factory=dict) @model_validator(mode="after") def validate_navigation(self) -> "NavigationBuildSpec": if self.navigation.enabled and not self.navigation.items: raise ValueError("navigation.items must be a non-empty list when navigation.enabled is true") return self BuildManifest = SolutionSpec def _validate_sample_value(entity_id: str, record_name: str, field: FieldSpec, raw_value: Any) -> None: field_type = field.type if raw_value is None: return if field_type == FieldType.number: if isinstance(raw_value, bool) or not isinstance(raw_value, (int, float)): raise ValueError( f"entity '{entity_id}' sample record '{record_name}' field '{field.field_id}' expects an integer-compatible number" ) if isinstance(raw_value, float) and not raw_value.is_integer(): raise ValueError( f"entity '{entity_id}' sample record '{record_name}' field '{field.field_id}' must be an integer-compatible number; use amount for decimals" ) return if field_type == FieldType.amount: if isinstance(raw_value, bool) or not isinstance(raw_value, (int, float)): raise ValueError(f"entity '{entity_id}' sample record '{record_name}' field '{field.field_id}' expects a numeric amount") return if field_type == FieldType.member: if isinstance(raw_value, int) and raw_value > 0: return if isinstance(raw_value, dict) and isinstance(raw_value.get("id") or raw_value.get("uid"), int): return raise ValueError(f"entity '{entity_id}' sample record '{record_name}' field '{field.field_id}' expects a positive member id or member object") if field_type == FieldType.boolean: if not isinstance(raw_value, bool): raise ValueError(f"entity '{entity_id}' sample record '{record_name}' field '{field.field_id}' expects a boolean") return if field_type == FieldType.single_select: if not isinstance(raw_value, str): raise ValueError(f"entity '{entity_id}' sample record '{record_name}' field '{field.field_id}' expects a string option") if field.options and raw_value not in field.options: raise ValueError( f"entity '{entity_id}' sample record '{record_name}' field '{field.field_id}' uses unknown option '{raw_value}'" ) return if field_type == FieldType.multi_select: items = raw_value if isinstance(raw_value, list) else [raw_value] if not all(isinstance(item, str) for item in items): raise ValueError(f"entity '{entity_id}' sample record '{record_name}' field '{field.field_id}' expects string options") if field.options: unknown = [item for item in items if item not in field.options] if unknown: raise ValueError( f"entity '{entity_id}' sample record '{record_name}' field '{field.field_id}' uses unknown options {unknown}" ) return if field_type in {FieldType.date, FieldType.datetime, FieldType.text, FieldType.long_text, FieldType.phone, FieldType.email, FieldType.address}: if not isinstance(raw_value, str): raise ValueError(f"entity '{entity_id}' sample record '{record_name}' field '{field.field_id}' expects a string") return if field_type == FieldType.relation: refs = raw_value if isinstance(raw_value, list) else [raw_value] if not all(isinstance(item, (str, int)) for item in refs): raise ValueError(f"entity '{entity_id}' sample record '{record_name}' field '{field.field_id}' expects record reference ids") return if field_type == FieldType.attachment: items = raw_value if isinstance(raw_value, list) else [raw_value] if not all(isinstance(item, dict) for item in items): raise ValueError(f"entity '{entity_id}' sample record '{record_name}' field '{field.field_id}' expects attachment objects") return FieldSpec.model_rebuild() NavigationItemSpec.model_rebuild()