Source code for quantumengine.fields.additional
from typing import Any, Dict, List, Optional, Type, Union
from .base import Field
[docs]
class OptionField(Field):
"""Option field type.
This field type makes a field optional and guarantees it to be either
None or a value of the specified type.
Attributes:
field_type: The field type for the value when not None
"""
[docs]
def __init__(self, field_type: Field, **kwargs: Any) -> None:
"""Initialize a new OptionField.
Args:
field_type: The field type for the value when not None
**kwargs: Additional arguments to pass to the parent class
"""
self.field_type = field_type
super().__init__(**kwargs)
self.py_type = Any
[docs]
def validate(self, value: Any) -> Any:
"""Validate the option value.
This method checks if the value is None or a valid value for the field_type.
Args:
value: The value to validate
Returns:
The validated value
Raises:
ValueError: If the value is not None and fails validation for field_type
"""
# Skip the parent's validation since we handle required differently
if value is None:
return None
return self.field_type.validate(value)
[docs]
def to_db(self, value: Any) -> Any:
"""Convert Python value to database representation.
This method converts a Python value to a database representation using
the field_type's to_db method if the value is not None.
Args:
value: The Python value to convert
Returns:
The database representation of the value
"""
if value is None:
return None
return self.field_type.to_db(value)
[docs]
def from_db(self, value: Any) -> Any:
"""Convert database value to Python representation.
This method converts a database value to a Python representation using
the field_type's from_db method if the value is not None.
Args:
value: The database value to convert
Returns:
The Python representation of the value
"""
if value is None:
return None
return self.field_type.from_db(value)
[docs]
class FutureField(Field):
"""Field for future (computed) values.
This field type represents a computed value in SurrealDB that is calculated
at query time rather than stored in the database. It uses SurrealDB's
<future> syntax to define a computation expression.
Attributes:
computation_expression: The SurrealDB expression to compute the value
"""
[docs]
def __init__(self, computation_expression: str, **kwargs: Any) -> None:
"""Initialize a new FutureField.
Args:
computation_expression: The SurrealDB expression to compute the value
**kwargs: Additional arguments to pass to the parent class
"""
self.computation_expression = computation_expression
super().__init__(**kwargs)
self.py_type = Any
[docs]
def to_db(self, value: Any) -> str:
"""Convert to SurrealDB future syntax.
This method returns the SurrealDB <future> syntax with the computation
expression, regardless of the input value.
Args:
value: The input value (ignored)
Returns:
The SurrealDB future syntax string
"""
# For future fields, we return a special SurrealDB syntax
return f"<future> {{ {self.computation_expression} }}"
[docs]
class TableField(Field):
"""Table field type.
This field type stores table names and provides validation and
conversion between Python strings and SurrealDB table format.
Example:
```python
class Schema(Document):
table_name = TableField()
fields = DictField()
```
"""
[docs]
def __init__(self, **kwargs: Any) -> None:
"""Initialize a new TableField.
Args:
**kwargs: Additional arguments to pass to the parent class
"""
super().__init__(**kwargs)
self.py_type = str
[docs]
def validate(self, value: Any) -> Optional[str]:
"""Validate the table name.
This method checks if the value is a valid table name.
Args:
value: The value to validate
Returns:
The validated table name
Raises:
TypeError: If the value is not a string
ValueError: If the table name is invalid
"""
value = super().validate(value)
if value is not None:
if not isinstance(value, str):
raise TypeError(f"Expected string for table name in field '{self.name}', got {type(value)}")
# Basic validation for table names
if not value or ' ' in value:
raise ValueError(f"Invalid table name '{value}' for field '{self.name}'")
return value
[docs]
def to_db(self, value: Any) -> Optional[str]:
"""Convert Python string to database representation.
This method converts a Python string to a table name for storage in the database.
Args:
value: The Python string to convert
Returns:
The table name for the database
"""
if value is not None and not isinstance(value, str):
try:
return str(value)
except (TypeError, ValueError):
pass
return value
[docs]
class RangeField(Field):
"""Field for storing ranges of values.
This field type stores ranges of values with minimum and maximum bounds.
It supports various types for the bounds, such as numbers, strings, and dates.
Example:
class PriceRange(Document):
price_range = RangeField(min_type=FloatField(), max_type=FloatField())
age_range = RangeField(min_type=IntField(), max_type=IntField())
"""
[docs]
def __init__(self, min_type: Field, max_type: Field = None, **kwargs: Any) -> None:
"""Initialize a new RangeField.
Args:
min_type: The field type for the minimum value
max_type: The field type for the maximum value (defaults to same as min_type)
**kwargs: Additional arguments to pass to the parent class
"""
self.min_type = min_type
self.max_type = max_type if max_type is not None else min_type
super().__init__(**kwargs)
self.py_type = Dict[str, Any]
[docs]
def validate(self, value: Any) -> Optional[Dict[str, Any]]:
"""Validate the range value.
This method checks if the value is a valid range with minimum and maximum
values that can be validated by the respective field types.
Args:
value: The value to validate
Returns:
The validated range value
Raises:
ValidationError: If the value is not a valid range
"""
value = super().validate(value)
if value is None:
return None
if not isinstance(value, dict):
from ..exceptions import ValidationError
raise ValidationError(f"Expected dict for field '{self.name}', got {type(value)}")
# Ensure the range has min and max keys
if 'min' not in value and 'max' not in value:
from ..exceptions import ValidationError
raise ValidationError(f"Range field '{self.name}' must have at least one of 'min' or 'max' keys")
# Validate min value if present
if 'min' in value:
try:
value['min'] = self.min_type.validate(value['min'])
except (TypeError, ValueError) as e:
from ..exceptions import ValidationError
raise ValidationError(f"Invalid minimum value for field '{self.name}': {str(e)}")
# Validate max value if present
if 'max' in value:
try:
value['max'] = self.max_type.validate(value['max'])
except (TypeError, ValueError) as e:
from ..exceptions import ValidationError
raise ValidationError(f"Invalid maximum value for field '{self.name}': {str(e)}")
# Ensure min <= max if both are present
if 'min' in value and 'max' in value:
min_val = value['min']
max_val = value['max']
# Skip comparison if either value is None
if min_val is not None and max_val is not None:
# Try to compare the values
try:
if min_val > max_val:
from ..exceptions import ValidationError
raise ValidationError(f"Minimum value ({min_val}) cannot be greater than maximum value ({max_val}) for field '{self.name}'")
except TypeError:
# If values can't be compared, just skip the check
pass
return value
[docs]
def to_db(self, value: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
"""Convert Python range to database representation.
Args:
value: The Python range to convert
Returns:
The database representation of the range
"""
if value is None:
return None
result = {}
# Convert min value if present
if 'min' in value and value['min'] is not None:
result['min'] = self.min_type.to_db(value['min'])
# Convert max value if present
if 'max' in value and value['max'] is not None:
result['max'] = self.max_type.to_db(value['max'])
return result
[docs]
def from_db(self, value: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
"""Convert database range to Python representation.
Args:
value: The database range to convert
Returns:
The Python representation of the range
"""
if value is None:
return None
result = {}
# Convert min value if present
if 'min' in value and value['min'] is not None:
result['min'] = self.min_type.from_db(value['min'])
# Convert max value if present
if 'max' in value and value['max'] is not None:
result['max'] = self.max_type.from_db(value['max'])
return result