Generate a Redis key based on the object name, primary key fields, and an optional postfix.
(self, postfix: str = None, **kwargs)
| 24 | # --- redis --- |
| 25 | |
| 26 | def __key(self, postfix: str = None, **kwargs) -> str: |
| 27 | """Generate a Redis key based on the object name, primary key fields, and an optional postfix.""" |
| 28 | # Start with the object name |
| 29 | key_parts = [self.entity_class.object_name()] |
| 30 | |
| 31 | # Add the primary key components in the specified order |
| 32 | for field in self.entity_class.primary_key_fields(): |
| 33 | if field in kwargs: |
| 34 | key_parts.append(str(kwargs[field])) |
| 35 | else: |
| 36 | raise ValueError(f"Missing value for primary key field: {field}") |
| 37 | |
| 38 | # If a postfix is provided, append it |
| 39 | if postfix: |
| 40 | key_parts.append(postfix) |
| 41 | |
| 42 | # Join all parts with a colon |
| 43 | return ":".join(key_parts) |
| 44 | |
| 45 | async def get(self, **kwargs) -> Optional[ModelEntity]: |
| 46 | if not self.redis_conn: |
no test coverage detected