Changeset bf30018 in flexoentity
- Timestamp:
- 11/04/25 08:05:04 (2 months ago)
- Branches:
- master
- Children:
- 4af65b0
- Parents:
- 5c72356
- Location:
- flexoentity
- Files:
-
- 2 edited
-
domain.py (modified) (4 diffs)
-
flexo_entity.py (modified) (24 diffs)
Legend:
- Unmodified
- Added
- Removed
-
flexoentity/domain.py
r5c72356 rbf30018 1 1 from dataclasses import dataclass 2 from flexoentity .flexo_entityimport FlexOID, FlexoEntity, EntityType, EntityState2 from flexoentity import FlexOID, FlexoEntity, EntityType, EntityState 3 3 4 4 5 5 @dataclass 6 6 class Domain(FlexoEntity): 7 """ 8 I am a helper class to provide more information than just a 9 domain abbreviation in FlexOID, doing mapping and management 10 """ 11 ENTITY_TYPE = EntityType.DOMAIN 12 7 13 fullname: str = "" 8 14 description: str = "" 9 15 classification: str = "UNCLASSIFIED" 10 owner: str = "unknown"11 16 12 17 @classmethod 13 18 def default(cls): 14 return cls(domain="GEN") 19 """I return an default instance of myself""" 20 return cls("GENERIC") 15 21 16 22 def __post_init__(self): … … 28 34 ) 29 35 30 31 32 36 @property 33 37 def text_seed(self) -> str: 34 """Deterministic text seed for ID generation.""" 35 return f"{self.fullname}|{self.classification}|{self.owner}" 38 """ 39 I provide a deterministic text seed for ID generation. 40 FIXME: There might be a better text seed, json probably 41 """ 42 return f"{self.fullname}|{self.classification}|{self.owner_id}" 36 43 37 44 def to_dict(self): 45 """I return a dictionary representing my state""" 38 46 base = super().to_dict() 39 47 base.update({ … … 41 49 "description": self.description, 42 50 "classification": self.classification, 43 "owner": self.owner44 51 }) 45 52 return base … … 47 54 @classmethod 48 55 def from_dict(cls, data): 56 """I return a new instance of myself, created from state provided by a dictionary""" 49 57 return cls( 50 58 fullname=data.get("fullname", ""), 51 59 description=data.get("description", ""), 52 60 classification=data.get("classification", "UNCLASSIFIED"), 53 owner=data.get("owner", "unknown"),54 61 ) -
flexoentity/flexo_entity.py
r5c72356 rbf30018 17 17 - state (single capital letter) 18 18 19 - My derived properties ( `state`, `entity_type`, `version`) read from that ID.19 - My derived properties (state, entity_type, version) read from that ID. 20 20 They are never stored separately, so I cannot become inconsistent. 21 21 22 - My attributes `origin`, `originator_id`, and `owner_id`express authorship22 - My attributes origin, originator_id, and owner_id express authorship 23 23 and workflow context: 24 - `origin`links to the FlexOID of the entity I was derived from25 - `originator_id`identifies the original creator (UUID)26 - `owner_id`identifies the current responsible user or process (UUID)27 28 - My optional `domain`field is a semantic hint, useful for filtering or24 - origin links to the FlexOID of the entity I was derived from 25 - originator_id identifies the original creator (UUID) 26 - owner_id identifies the current responsible user or process (UUID) 27 28 - My domain field is a semantic hint, useful for filtering or 29 29 multi-tenant setups, but the canonical domain remains encoded in my ID. 30 30 … … 49 49 - My identity never changes; only my content or ownership does. 50 50 - My lifecycle transitions (approve, sign, publish) are represented by new 51 FlexOIDs, not by mutating stateattributes.51 FlexOIDs, not by mutating my own attributes. 52 52 - My audit trail and access control live in the provenance layer, not the ID. 53 53 … … 73 73 74 74 class EntityType(Enum): 75 """ 76 I map a fixed number of entity types to a single unique letter, 77 that structures different types 78 """ 75 79 GENERIC = "G" 76 80 DOMAIN = "D" … … 89 93 @classmethod 90 94 def from_letter(cls, a_letter): 95 """I am a convenience constructor method for EntityType(a_letter)""" 91 96 return cls(a_letter) 92 93 # FIXME: Add more mappings94 def short(self) -> str:95 mapping = {96 EntityType.MEDIA: "M",97 EntityType.DOMAIN: "DOM",98 EntityType.CATALOG: "CAT",99 }100 return mapping[self]101 97 102 98 # ────────────────────────────────────────────────────────────────────────────── … … 106 102 107 103 class EntityState(Enum): 104 """I describe the life cycle states for all entity types""" 108 105 DRAFT = "D" 109 106 APPROVED = "A" … … 116 113 117 114 118 @dataclass( kw_only=True)115 @dataclass() 119 116 class FlexoEntity(ABC): 120 117 """ 121 118 I represent a mutable entity identified by an immutable FlexOID. 122 119 123 My `flexo_id`is the single source of truth for what I am:120 My flexo_id is the single source of truth for what I am: 124 121 - it encodes my domain, entity type, version, and lifecycle state. 125 122 I never store those values separately, but derive them on demand. 126 123 127 124 I extend the raw identity with provenance and authorship data: 128 - `origin` → the FlexOID of the entity I was derived from 129 - `originator_id` → the UUID of the first creator 130 - `owner_id` → the UUID of the current responsible person or system 131 - `domain` → optional semantic domain hint (for filtering or indexing) 125 - origin -> the FlexOID of the entity I was derived from 126 - originator_id -> the UUID of the first creator 127 - owner_id -> the UUID of the current responsible person or system 128 - domain -> FIXME: Explain reasons, why keep domain mutable. semantic 129 domain hint (for filtering or indexing) 132 130 133 131 I am designed to be traceable and reproducible: 134 - My identity ( `flexo_id`) never mutates.132 - My identity (flexo_id) never mutates. 135 133 - My lifecycle transitions are represented by new FlexOIDs. 136 134 - My provenance (origin, originator_id, owner_id) can change over time … … 139 137 Behavior 140 138 ───────── 141 - `state` and `entity_type`are read-only views derived from my FlexOID.142 - `to_dict()` and `from_dict()`serialize me with human-readable fields139 - state and entity_type are read-only views derived from my FlexOID. 140 - to_dict() and from_dict() serialize me with human-readable fields 143 141 while preserving the canonical FlexOID. 144 142 - Subclasses (such as questions, media items, or catalogs) extend me with 145 143 domain-specific content and validation, but not with new identity rules. 144 - Subclasses have to define a class attribute ENTITY_TYPE to allow specific instance creation 146 145 147 146 I am the living, editable layer that connects identity with accountability. … … 157 156 158 157 def with_new_owner(self, new_owner: UUID): 159 """ Return a clone of this entity with a different owner."""158 """I return a clone of this entity with a different owner.""" 160 159 copy = replace(self) 161 160 copy.owner_id = new_owner … … 165 164 @abstractmethod 166 165 def text_seed(self) -> str: 167 """ Canonicalized text used for ID generation."""166 """I provide a canonicalized text used for ID generation.""" 168 167 raise NotImplementedError("Subclasses must define text_seed property") 169 168 170 169 @property 171 170 def state(self) -> EntityState: 171 """I return the state derived from my FlexOID""" 172 172 return EntityState(self.flexo_id.state_code) 173 173 174 174 @property 175 175 def entity_type(self) -> EntityType: 176 return EntityType(self.flexo_id.entity_type) 176 """I return the entity type derived from my FlexOID""" 177 return EntityType(self.flexo_id.entity_type) 177 178 178 179 def canonical_seed(self) -> str: 180 """ 181 I use a helper method to flatten my text_seed. 182 NOTE:This might become superfluous when text_seed provides this by itself. Redesign possible 183 """ 179 184 return canonical_seed(self.text_seed) 180 185 … … 182 187 @abstractmethod 183 188 def default(cls): 184 """ Return a minimal valid instance of this entity (DRAFT state)."""189 """I return a minimal valid instance of this entity (in DRAFT state).""" 185 190 raise NotImplementedError("Subclasses must implement default()") 186 191 187 192 def domain_code(self) -> str: 188 """Return canonical domain code for serialization and ID generation.""" 193 """ 194 I return a canonical domain code for serialization and ID generation. 195 As domains have their own domain code, I need to distinguish between Domains and 196 other FlexoEntities to avoid cyclic structures. 197 FIXME: Write a better comment 198 """ 189 199 return self.domain.domain if hasattr(self.domain, "domain") else self.domain 190 200 … … 193 203 Optionally generate a new FlexOID and fingerprint if none exist. 194 204 195 I check for subclass attribute s `ENTITY_TYPE` and `INITIAL_STATE`.196 If they exist, I use themto generate a draft FlexOID.205 I check for subclass attribute ENTITY_TYPE. 206 If it exist, I use it to generate a draft FlexOID. 197 207 Otherwise, I skip ID generation — leaving it to the subclass. 198 208 """ … … 202 212 return 203 213 204 # Skip if subclass doesn’t declare ENTITY_TYPE / INITIAL_STATE214 # Skip if subclass doesn’t declare ENTITY_TYPE or initial state 205 215 etype = getattr(self.__class__, "ENTITY_TYPE", None) 206 216 if not etype: … … 210 220 # Generate a new draft FlexOID 211 221 self.flexo_id = FlexOID.safe_generate( 212 self.domain_code(),213 e type.value,214 EntityState.DRAFT.value,215 self.text_seed,216 1,222 domain=self.domain_code(), 223 entity_type=etype.value, 224 estate=EntityState.DRAFT.value, 225 text=self.text_seed, 226 version=1, 217 227 ) 218 228 … … 234 244 "fingerprint": self.fingerprint, 235 245 "origin": self.origin, 246 "originator_id": str(self.originator_id), 247 "owner_id": str(self.owner_id) 236 248 } 237 249 … … 239 251 def from_dict(cls, data): 240 252 from flexoentity.domain import Domain # avoid circular import 241 domain = data["domain"] 242 abbrev, fullname = (lambda p: (p[0], p[1] if len(p) > 1 else ""))(domain.split("_", 1)) 243 domain_obj = Domain( 244 domain=abbrev 245 ) 253 domain_obj = Domain(data.get("domain", "GENERIC")) 246 254 obj = cls( 247 255 domain=domain_obj, … … 250 258 obj.fingerprint = data.get("fingerprint", "") 251 259 obj.origin = data.get("origin") 260 obj.originator_id = UUID(int=int(data.get("originator_id", "0"))) 261 obj.owner_id = UUID(int=int(data.get("owner_id", "0"))) 252 262 return obj 253 263 254 264 def to_json(self, *, indent: int | None = None) -> str: 255 """ Serialize entity (and itsFlexOID) into JSON."""265 """I serialize myself (and my FlexOID) into JSON.""" 256 266 return json.dumps(self.to_dict(), indent=indent, ensure_ascii=False) 257 267 258 268 @classmethod 259 269 def from_json(cls, data_str: str) -> "FlexoEntity": 260 """ Deserialize from a JSON string."""270 """I create a new instance from a JSON string.""" 261 271 data = json.loads(data_str) 262 272 return cls.from_dict(data) … … 265 275 def should_version(state) -> bool: 266 276 """ 267 Determine if a given lifecycle state should trigger a version increment. 277 FIXME: This is now superfluous, because lifecycle changes should not increment 278 version at all. 279 I determine if a given lifecycle state should trigger a version increment. 268 280 269 281 Entities typically version when they move into more stable or … … 277 289 ) 278 290 def _compute_fingerprint(self) -> str: 279 """ Always recompute the entity's content fingerprint."""291 """I always recompute the entity's content fingerprint.""" 280 292 seed = self.canonical_seed() 281 293 return hashlib.blake2s(seed.encode("utf-8"), digest_size=8).hexdigest().upper() 282 294 283 295 def _update_fingerprint(self) -> bool: 284 """Update FlexOID if the content fingerprint changed.""" 296 """ 297 FIXME: I am not sure if this is correct behaviour. 298 I Update FlexOID if the content fingerprint changed. 299 """ 285 300 new_fp = self._compute_fingerprint() 286 301 if new_fp != self.fingerprint: … … 296 311 def _transition(self, target_state: EntityState): 297 312 """ 298 I nternal helper for state transitions with version/fingerprint checks313 I am an internal helper for state transitions with version/fingerprint checks 299 314 and forward-only enforcement. 300 315 """ … … 326 341 @property 327 342 def version(self) -> int: 328 """ Extract numeric version from the FlexO-ID string."""343 """I extract numeric version from the FlexO-ID string.""" 329 344 try: 330 345 return int(str(self.flexo_id).rsplit("@", 1)[-1]) … … 333 348 334 349 def bump_version(self): 335 """I ncrementversion number on the ID."""350 """I increment the version number on the ID.""" 336 351 self.flexo_id = FlexOID.next_version(self.flexo_id) 337 352 338 353 def lineage(self, repo): 339 """ Return full ancestry chain [origin → ... → self]."""354 """I return full ancestry chain [origin → ... → self].""" 340 355 chain = [self] 341 356 current = self … … 347 362 def approve(self): 348 363 """ 349 Move from DRAFT to APPROVED state.364 I move from DRAFT to APPROVED state. 350 365 Draft entities receive a new permanent FlexOID with incremented version. 351 366 """ 352 353 367 if self.state == EntityState.DRAFT: 354 368 new_fid = FlexOID.safe_generate(self.domain_code(), … … 366 380 def sign(self): 367 381 """ 368 Mark entity as approved and signed without bumping version.382 I mark entity as approved and signed without bumping version. 369 383 Only changes the state letter in the FlexOID. 384 FIXME: We need a signature here 370 385 """ 371 386 if self.state != EntityState.APPROVED: … … 378 393 def publish(self): 379 394 """ 380 Move from APPROVED or APPROVED_AND_SIGNED to PUBLISHED.395 I move from APPROVED or APPROVED_AND_SIGNED to PUBLISHED. 381 396 382 397 Uses allowed_transitions() to verify legality,
Note:
See TracChangeset
for help on using the changeset viewer.
