Coverage for src/dataknobs_data/records.py: 45%
213 statements
« prev ^ index » next coverage.py v7.10.3, created at 2025-08-17 19:59 -0500
1import uuid
2from collections import OrderedDict
3from collections.abc import Iterator
4from dataclasses import dataclass, field
5from typing import Any
7from .fields import Field, FieldType
10@dataclass
11class Record:
12 """Represents a structured data record with fields and metadata.
14 The record ID can be accessed via the `id` property, which:
15 - Returns the explicitly set ID if available
16 - Falls back to metadata['id'] if present
17 - Returns None if no ID is set
18 """
20 fields: OrderedDict[str, Field] = field(default_factory=OrderedDict)
21 metadata: dict[str, Any] = field(default_factory=dict)
22 _id: str | None = field(default=None, repr=False)
24 def __init__(
25 self,
26 data: dict[str, Any] | OrderedDict[str, Field] | None = None,
27 metadata: dict[str, Any] | None = None,
28 id: str | None = None,
29 ):
30 """Initialize a record from various data formats.
32 Args:
33 data: Can be a dict of field names to values, or an OrderedDict of Field objects
34 metadata: Optional metadata for the record
35 id: Optional unique identifier for the record
36 """
37 self.metadata = metadata or {}
38 self.fields = OrderedDict()
39 self._id = id
41 # Process data first to populate fields
42 if data:
43 if isinstance(data, OrderedDict) and all(
44 isinstance(v, Field) for v in data.values()
45 ):
46 self.fields = data
47 else:
48 for key, value in data.items():
49 if isinstance(value, Field):
50 self.fields[key] = value
51 else:
52 self.fields[key] = Field(name=key, value=value)
54 # Now check for ID from various sources if not explicitly provided
55 if self._id is None:
56 # Check metadata
57 if "id" in self.metadata:
58 self._id = str(self.metadata["id"])
59 # Check fields for id
60 elif "id" in self.fields:
61 value = self.get_value("id")
62 if value is not None:
63 self._id = str(value)
64 # Sync to metadata
65 self.metadata["id"] = self._id
66 # Check fields for record_id
67 elif "record_id" in self.fields:
68 value = self.get_value("record_id")
69 if value is not None:
70 self._id = str(value)
71 # Sync to metadata
72 self.metadata["id"] = self._id
74 @property
75 def id(self) -> str | None:
76 """Get the record ID.
78 Checks for ID in the following priority order:
79 1. Explicitly set ID (_id)
80 2. ID in metadata
81 3. ID field in record fields
82 4. record_id field in record fields (common in DataFrames)
84 Returns the first ID found, or None if no ID is present.
85 """
86 # 1. Check explicitly set ID
87 if self._id is not None:
88 return self._id
90 # 2. Check metadata
91 if "id" in self.metadata:
92 return str(self.metadata["id"])
94 # 3. Check for 'id' field
95 if "id" in self.fields:
96 value = self.get_value("id")
97 if value is not None:
98 return str(value)
100 # 4. Check for 'record_id' field (common in DataFrames)
101 if "record_id" in self.fields:
102 value = self.get_value("record_id")
103 if value is not None:
104 return str(value)
106 return None
108 @id.setter
109 def id(self, value: str | None) -> None:
110 """Set the record ID.
112 Updates the ID in all locations for consistency:
113 - Internal _id attribute
114 - Metadata (for backward compatibility)
115 - ID field if it exists
116 """
117 self._id = value
119 # Update metadata for backward compatibility
120 if value is not None:
121 self.metadata["id"] = value
122 elif "id" in self.metadata:
123 del self.metadata["id"]
125 # Update ID field if it exists (don't create it if it doesn't)
126 if "id" in self.fields and value is not None:
127 self.fields["id"].value = value
129 # Update record_id field if it exists (common in DataFrames)
130 if "record_id" in self.fields and value is not None:
131 self.fields["record_id"].value = value
133 def generate_id(self) -> str:
134 """Generate and set a new UUID for this record.
136 Returns:
137 The generated UUID string
138 """
139 new_id = str(uuid.uuid4())
140 self.id = new_id
141 return new_id
143 def get_field(self, name: str) -> Field | None:
144 """Get a field by name."""
145 return self.fields.get(name)
147 def get_value(self, name: str, default: Any = None) -> Any:
148 """Get a field's value by name, supporting dot-notation for nested paths.
150 Args:
151 name: Field name or dot-notation path (e.g., "metadata.type")
152 default: Default value if field not found
154 Returns:
155 The field value or default
156 """
157 # Check if this is a nested path
158 if "." in name:
159 return self.get_nested_value(name, default)
161 # Simple field lookup
162 field = self.get_field(name)
163 return field.value if field else default
165 def get_nested_value(self, path: str, default: Any = None) -> Any:
166 """Get a value from a nested path using dot notation.
168 Supports paths like:
169 - "metadata.type" - access metadata dict
170 - "fields.temperature" - access field values
171 - "metadata.config.timeout" - nested dict access
173 Args:
174 path: Dot-notation path to the value
175 default: Default value if path not found
177 Returns:
178 The value at the path or default
179 """
180 parts = path.split(".", 1)
181 if len(parts) == 1:
182 # No more nesting, get the value
183 return self.get_value(parts[0], default)
185 root, remaining = parts
187 # Handle special root paths
188 if root == "metadata":
189 # Navigate through metadata dict
190 if not self.metadata:
191 return default
192 return self._traverse_dict(self.metadata, remaining, default)
193 elif root == "fields":
194 # Get field value by name
195 if "." in remaining:
196 # Nested path within field value (if it's a dict)
197 field_name, field_path = remaining.split(".", 1)
198 field_value = self.get_value(field_name, None)
199 if isinstance(field_value, dict):
200 return self._traverse_dict(field_value, field_path, default)
201 return default
202 else:
203 # Simple field access
204 return self.get_value(remaining, default)
205 else:
206 # Check if it's a field containing a dict
207 field_value = self.get_value(root, None)
208 if isinstance(field_value, dict):
209 return self._traverse_dict(field_value, remaining, default)
210 return default
212 def _traverse_dict(self, data: dict, path: str, default: Any = None) -> Any:
213 """Traverse a dictionary using dot notation.
215 Args:
216 data: Dictionary to traverse
217 path: Dot-notation path
218 default: Default value if path not found
220 Returns:
221 Value at path or default
222 """
223 parts = path.split(".")
224 current = data
226 for part in parts:
227 if isinstance(current, dict) and part in current:
228 current = current[part]
229 else:
230 return default
232 return current
234 def set_field(
235 self,
236 name: str,
237 value: Any,
238 field_type: FieldType | None = None,
239 field_metadata: dict[str, Any] | None = None,
240 ) -> None:
241 """Set or update a field."""
242 self.fields[name] = Field(
243 name=name, value=value, type=field_type, metadata=field_metadata or {}
244 )
246 def remove_field(self, name: str) -> bool:
247 """Remove a field by name. Returns True if field was removed."""
248 if name in self.fields:
249 del self.fields[name]
250 return True
251 return False
253 def has_field(self, name: str) -> bool:
254 """Check if a field exists."""
255 return name in self.fields
257 def field_names(self) -> list[str]:
258 """Get list of field names."""
259 return list(self.fields.keys())
261 def field_count(self) -> int:
262 """Get the number of fields."""
263 return len(self.fields)
265 def __getitem__(self, key: str | int) -> Any:
266 """Get field value by name or field by index.
268 For string keys, returns the field value directly (dict-like access).
269 For integer keys, returns the Field object at that index for backward compatibility.
270 """
271 if isinstance(key, str):
272 if key not in self.fields:
273 raise KeyError(f"Field '{key}' not found")
274 return self.fields[key].value
275 elif isinstance(key, int):
276 field_list = list(self.fields.values())
277 if key < 0 or key >= len(field_list):
278 raise IndexError(f"Field index {key} out of range")
279 return field_list[key]
280 else:
281 raise TypeError(f"Key must be str or int, got {type(key)}")
283 def __setitem__(self, key: str, value: Field | Any) -> None:
284 """Set field by name.
286 Can accept either a Field object or a raw value.
287 When given a raw value, creates a new Field automatically.
288 """
289 if isinstance(value, Field):
290 self.fields[key] = value
291 else:
292 self.set_field(key, value)
294 def __delitem__(self, key: str) -> None:
295 """Delete field by name."""
296 if key not in self.fields:
297 raise KeyError(f"Field '{key}' not found")
298 del self.fields[key]
300 def __contains__(self, key: str) -> bool:
301 """Check if field exists."""
302 return key in self.fields
304 def __iter__(self) -> Iterator[str]:
305 """Iterate over field names."""
306 return iter(self.fields)
308 def __len__(self) -> int:
309 """Get number of fields."""
310 return len(self.fields)
312 def validate(self) -> bool:
313 """Validate all fields in the record."""
314 return all(field.validate() for field in self.fields.values())
316 def get_field_object(self, key: str) -> Field:
317 """Get the Field object by name.
319 Use this method when you need access to the Field object itself,
320 not just its value.
322 Args:
323 key: Field name
325 Returns:
326 The Field object
328 Raises:
329 KeyError: If field not found
330 """
331 if key not in self.fields:
332 raise KeyError(f"Field '{key}' not found")
333 return self.fields[key]
335 def __getattr__(self, name: str) -> Any:
336 """Get field value by attribute access.
338 Provides convenient attribute-style access to field values.
339 Falls back to normal attribute access for non-field attributes.
341 Args:
342 name: Attribute/field name
344 Returns:
345 Field value if field exists, otherwise raises AttributeError
346 """
347 # Avoid infinite recursion for special attributes
348 if name.startswith("_") or name in ("fields", "metadata", "id"):
349 raise AttributeError(
350 f"'{type(self).__name__}' object has no attribute '{name}'"
351 )
353 # Check if it's a field
354 if hasattr(self, "fields") and name in self.fields:
355 return self.fields[name].value
357 raise AttributeError(f"'{type(self).__name__}' object has no field '{name}'")
359 def __setattr__(self, name: str, value: Any) -> None:
360 """Set field value by attribute access.
362 Allows setting field values using attribute syntax.
363 Special attributes (fields, metadata, _id) are handled normally.
365 Args:
366 name: Attribute/field name
367 value: Value to set
368 """
369 # Handle special attributes normally
370 if name in ("fields", "metadata", "_id") or name.startswith("_"):
371 super().__setattr__(name, value)
372 elif hasattr(self, "fields") and name in self.fields:
373 # Update existing field value
374 self.fields[name].value = value
375 else:
376 # For new fields during normal operation, create them
377 # But during __init__, we need to use normal attribute setting
378 if hasattr(self, "fields"):
379 self.set_field(name, value)
380 else:
381 super().__setattr__(name, value)
383 def to_dict(
384 self,
385 include_metadata: bool = False,
386 flatten: bool = True,
387 include_field_objects: bool = True,
388 ) -> dict[str, Any]:
389 """Convert record to dictionary.
391 Args:
392 include_metadata: Whether to include metadata in the output
393 flatten: If True (default), return just field values; if False, return structured format
394 include_field_objects: If True and not flattened, return full Field objects
396 Returns:
397 Dictionary representation of the record
398 """
399 if flatten:
400 # Simple dict with just values (default behavior for ergonomics)
401 result = {name: field.value for name, field in self.fields.items()}
402 if self.id:
403 result["_id"] = self.id
404 if include_metadata and self.metadata:
405 result["_metadata"] = self.metadata
406 else:
407 # Structured format for serialization
408 if include_field_objects:
409 result = {
410 "fields": {
411 name: field.to_dict() for name, field in self.fields.items()
412 }
413 }
414 else:
415 result = {
416 "fields": {name: field.value for name, field in self.fields.items()}
417 }
418 if self.id:
419 result["id"] = self.id
420 if include_metadata:
421 result["metadata"] = self.metadata
422 return result
424 @classmethod
425 def from_dict(cls, data: dict[str, Any]) -> "Record":
426 """Create a record from a dictionary representation."""
427 if "fields" in data:
428 fields = OrderedDict()
429 for name, field_data in data["fields"].items():
430 if isinstance(field_data, dict) and "value" in field_data:
431 fields[name] = Field.from_dict(field_data)
432 else:
433 fields[name] = Field(name=name, value=field_data)
434 metadata = data.get("metadata", {})
435 record_id = data.get("id") or data.get("_id")
436 return cls(data=fields, metadata=metadata, id=record_id)
437 else:
438 # Check for _id in flattened format
439 record_id = data.pop("_id", None) if "_id" in data else None
440 return cls(data=data, id=record_id)
442 def copy(self, deep: bool = True) -> "Record":
443 """Create a copy of the record.
445 Args:
446 deep: If True, create deep copies of fields and metadata
447 """
448 if deep:
449 import copy
451 new_fields = OrderedDict()
452 for name, field in self.fields.items():
453 new_fields[name] = Field(
454 name=field.name,
455 value=copy.deepcopy(field.value),
456 type=field.type,
457 metadata=copy.deepcopy(field.metadata),
458 )
459 new_metadata = copy.deepcopy(self.metadata)
460 else:
461 new_fields = OrderedDict(self.fields)
462 new_metadata = self.metadata.copy()
464 return Record(data=new_fields, metadata=new_metadata, id=self.id)
466 def project(self, field_names: list[str]) -> "Record":
467 """Create a new record with only specified fields."""
468 projected_fields = OrderedDict()
469 for name in field_names:
470 if name in self.fields:
471 projected_fields[name] = self.fields[name]
472 return Record(data=projected_fields, metadata=self.metadata.copy(), id=self.id)
474 def merge(self, other: "Record", overwrite: bool = True) -> "Record":
475 """Merge another record into this one.
477 Args:
478 other: The record to merge
479 overwrite: If True, overwrite existing fields; if False, keep existing
481 Returns:
482 A new merged record
483 """
484 merged_fields = OrderedDict(self.fields)
485 for name, field_obj in other.fields.items():
486 if overwrite or name not in merged_fields:
487 merged_fields[name] = field_obj
489 merged_metadata = self.metadata.copy()
490 if overwrite:
491 merged_metadata.update(other.metadata)
493 # Use the ID from this record, or from other if this doesn't have one
494 merged_id = self.id if self.id else other.id
496 return Record(data=merged_fields, metadata=merged_metadata, id=merged_id)