Custom Fields¶
Create specialized field types for your application.
Overview¶
Aksara's field system is extensible. You can create custom fields for:
- Domain-specific data types (phone numbers, currencies)
- Complex validation
- Custom serialization
- Database-specific types
Basic Custom Field¶
Field Structure¶
from aksara.fields import Field
from aksara.exceptions import ValidationError
class PercentageField(Field):
"""Field for percentage values (0-100)."""
def __init__(self, min_value=0, max_value=100, **kwargs):
self.min_value = min_value
self.max_value = max_value
super().__init__(**kwargs)
def validate(self, value):
"""Validate the value."""
if value is None and self.null:
return None
if not isinstance(value, (int, float)):
raise ValidationError("Must be a number")
if value < self.min_value or value > self.max_value:
raise ValidationError(
f"Must be between {self.min_value} and {self.max_value}"
)
return float(value)
def to_db(self, value):
"""Convert to database representation."""
if value is None:
return None
return float(value)
def from_db(self, value):
"""Convert from database representation."""
if value is None:
return None
return float(value)
def get_db_type(self):
"""Return the database column type."""
return "REAL"
Using the Field¶
class Product(Model):
name = fields.String(max_length=100)
discount = PercentageField(default=0)
# Usage
product = Product(name="Widget", discount=25.5)
await product.save()
Field Methods¶
Required Methods¶
| Method | Purpose |
|---|---|
validate(value) |
Validate and clean the value |
to_db(value) |
Convert Python → Database |
from_db(value) |
Convert Database → Python |
get_db_type() |
Database column type |
Optional Methods¶
| Method | Purpose |
|---|---|
get_default() |
Return the default value |
contribute_to_class(cls, name) |
Called when field is added to model |
deconstruct() |
For migrations |
__get__ / __set__ |
Descriptor protocol |
Practical Examples¶
Phone Number Field¶
import re
from aksara.fields import Field
from aksara.exceptions import ValidationError
class PhoneField(Field):
"""Phone number field with validation."""
PHONE_REGEX = re.compile(r'^\+?1?\d{9,15}$')
def __init__(self, region="US", **kwargs):
self.region = region
super().__init__(**kwargs)
def validate(self, value):
if value is None:
if self.null:
return None
raise ValidationError("Phone number is required")
# Normalize: remove spaces, dashes, parentheses
normalized = re.sub(r'[\s\-\(\)]', '', str(value))
if not self.PHONE_REGEX.match(normalized):
raise ValidationError("Invalid phone number format")
return normalized
def to_db(self, value):
return value
def from_db(self, value):
return value
def get_db_type(self):
return "VARCHAR(20)"
# Usage
class Contact(Model):
name = fields.String(max_length=100)
phone = PhoneField()
alt_phone = PhoneField(null=True)
Money Field¶
from decimal import Decimal
from aksara.fields import Field
from aksara.exceptions import ValidationError
class MoneyField(Field):
"""Field for monetary values."""
def __init__(self, currency="USD", max_digits=10, decimal_places=2, **kwargs):
self.currency = currency
self.max_digits = max_digits
self.decimal_places = decimal_places
super().__init__(**kwargs)
def validate(self, value):
if value is None:
if self.null:
return None
raise ValidationError("Value is required")
try:
decimal_value = Decimal(str(value))
except:
raise ValidationError("Invalid monetary value")
if decimal_value < 0:
raise ValidationError("Value cannot be negative")
# Round to decimal places
return decimal_value.quantize(Decimal(10) ** -self.decimal_places)
def to_db(self, value):
if value is None:
return None
# Store as integer cents for precision
return int(value * (10 ** self.decimal_places))
def from_db(self, value):
if value is None:
return None
return Decimal(value) / (10 ** self.decimal_places)
def get_db_type(self):
return "BIGINT"
# Usage
class Order(Model):
total = MoneyField(currency="USD")
tax = MoneyField(currency="USD", default=Decimal("0.00"))
Encrypted Field¶
from cryptography.fernet import Fernet
from aksara.fields import Field
from aksara.conf import settings
class EncryptedField(Field):
"""Field that encrypts data at rest."""
def __init__(self, **kwargs):
self._fernet = None
super().__init__(**kwargs)
@property
def fernet(self):
if self._fernet is None:
key = settings.ENCRYPTION_KEY.encode()
self._fernet = Fernet(key)
return self._fernet
def validate(self, value):
if value is None and self.null:
return None
return str(value)
def to_db(self, value):
if value is None:
return None
return self.fernet.encrypt(value.encode()).decode()
def from_db(self, value):
if value is None:
return None
return self.fernet.decrypt(value.encode()).decode()
def get_db_type(self):
return "TEXT"
# Usage
class User(Model):
email = fields.Email()
ssn = EncryptedField() # Encrypted at rest
Enum Field¶
from enum import Enum
from aksara.fields import Field
from aksara.exceptions import ValidationError
class EnumField(Field):
"""Field for Python enums."""
def __init__(self, enum_class, **kwargs):
self.enum_class = enum_class
super().__init__(**kwargs)
def validate(self, value):
if value is None:
if self.null:
return None
raise ValidationError("Value is required")
# Accept enum member or string value
if isinstance(value, self.enum_class):
return value
try:
return self.enum_class(value)
except ValueError:
valid = [e.value for e in self.enum_class]
raise ValidationError(f"Must be one of: {valid}")
def to_db(self, value):
if value is None:
return None
return value.value
def from_db(self, value):
if value is None:
return None
return self.enum_class(value)
def get_db_type(self):
return "VARCHAR(50)"
# Usage
class Status(Enum):
PENDING = "pending"
ACTIVE = "active"
COMPLETED = "completed"
class Task(Model):
title = fields.String(max_length=200)
status = EnumField(Status, default=Status.PENDING)
Slug Field¶
from slugify import slugify
from aksara.fields import Field
from aksara.exceptions import ValidationError
class SlugField(Field):
"""Auto-generating slug field."""
def __init__(self, source_field=None, **kwargs):
self.source_field = source_field
kwargs.setdefault("max_length", 200)
super().__init__(**kwargs)
def validate(self, value):
if value is None and self.null:
return None
if value:
return slugify(str(value))
return value
def contribute_to_class(self, cls, name):
super().contribute_to_class(cls, name)
# Auto-generate from source field
if self.source_field:
from aksara.signals import pre_save
@pre_save(cls)
async def auto_slug(sender, instance, **kwargs):
if not getattr(instance, name):
source_value = getattr(instance, self.source_field, "")
setattr(instance, name, slugify(source_value))
def to_db(self, value):
return value
def from_db(self, value):
return value
def get_db_type(self):
return f"VARCHAR({self.max_length})"
# Usage
class Post(Model):
title = fields.String(max_length=200)
slug = SlugField(source_field="title", unique=True)
Array Field (PostgreSQL)¶
import json
from aksara.fields import Field
from aksara.exceptions import ValidationError
class ArrayField(Field):
"""PostgreSQL array field."""
def __init__(self, base_type="text", **kwargs):
self.base_type = base_type
super().__init__(**kwargs)
def validate(self, value):
if value is None:
if self.null:
return None
raise ValidationError("Value is required")
if not isinstance(value, (list, tuple)):
raise ValidationError("Must be a list")
return list(value)
def to_db(self, value):
if value is None:
return None
# PostgreSQL array format: {item1,item2}
return "{" + ",".join(str(v) for v in value) + "}"
def from_db(self, value):
if value is None:
return None
if isinstance(value, list):
return value
# Parse PostgreSQL array format
return value.strip("{}").split(",") if value else []
def get_db_type(self):
return f"{self.base_type.upper()}[]"
# Usage
class Article(Model):
title = fields.String(max_length=200)
tags = ArrayField(base_type="text", default=list)
Field with Serialization¶
Custom Serialization¶
from aksara.fields import Field
from aksara.api.serializers import SerializerField
class ColorField(Field):
"""RGB color field."""
def validate(self, value):
if value is None and self.null:
return None
if isinstance(value, dict):
# {r: 255, g: 128, b: 0}
return value
if isinstance(value, str) and value.startswith("#"):
# #ff8000
return self.hex_to_rgb(value)
raise ValidationError("Invalid color format")
def hex_to_rgb(self, hex_color):
hex_color = hex_color.lstrip("#")
return {
"r": int(hex_color[0:2], 16),
"g": int(hex_color[2:4], 16),
"b": int(hex_color[4:6], 16),
}
def to_db(self, value):
if value is None:
return None
return f"{value['r']},{value['g']},{value['b']}"
def from_db(self, value):
if value is None:
return None
r, g, b = map(int, value.split(","))
return {"r": r, "g": g, "b": b}
def get_db_type(self):
return "VARCHAR(20)"
def get_serializer_field(self):
"""Return the API serializer field."""
return ColorSerializerField()
class ColorSerializerField(SerializerField):
"""Serializer field for colors."""
def to_representation(self, value):
"""Convert to JSON output."""
if value is None:
return None
return {
"rgb": value,
"hex": "#{r:02x}{g:02x}{b:02x}".format(**value),
}
def to_internal_value(self, data):
"""Convert from JSON input."""
if isinstance(data, str):
return {"hex": data}
return data
Migration Support¶
Deconstruction¶
For migrations to work, fields must be deconstructable:
class CustomField(Field):
def __init__(self, custom_arg, **kwargs):
self.custom_arg = custom_arg
super().__init__(**kwargs)
def deconstruct(self):
name, path, args, kwargs = super().deconstruct()
kwargs["custom_arg"] = self.custom_arg
return name, path, args, kwargs
Type Hints¶
Add Type Hints¶
Create a .pyi stub file:
# fields.pyi
from typing import TypeVar, Generic, Optional
T = TypeVar("T")
class PercentageField(Field[float]):
def __init__(
self,
min_value: float = 0,
max_value: float = 100,
null: bool = False,
default: Optional[float] = None,
) -> None: ...
Testing Custom Fields¶
import pytest
from aksara.testing import AksaraTestCase
class TestPercentageField(AksaraTestCase):
async def test_valid_value(self):
field = PercentageField()
assert field.validate(50) == 50.0
async def test_boundary_values(self):
field = PercentageField()
assert field.validate(0) == 0.0
assert field.validate(100) == 100.0
async def test_invalid_range(self):
field = PercentageField()
with pytest.raises(ValidationError):
field.validate(150)
async def test_custom_range(self):
field = PercentageField(min_value=10, max_value=50)
assert field.validate(30) == 30.0
with pytest.raises(ValidationError):
field.validate(5)