SQL Alchemy: Nested Composite Column Types and Custom Types

Question

I'm relatively new to SQLAlchemy and had a question regarding thoughts and advice on some approaches to composite and custom types.

Challenge: I use Value Objects quite heavily, with some value objects having attributes made up of Value Objects. ie.

class Customer:
name = CustomerName(first_name, last_name)
country = Country(country_name, region)

where Customer is an entity. CustomerName is a value object with primitive types (first_name, last_name), and Country is a value object made of of value objects (CountryName, Region).

I want the value objects to all persist in a single Customer table. For something like CustomerName, I can easily use a composite type:

class Customer(Base)
first_name = Column(String)
last_name = Column(String)
customer_name = composite(CustomerName, first_name, last_name)

Easy.

I have found that you can't nest composite types though (this may be possible in Postgres using sqlalchemy-utils but that is not an option for me).

I'm interested in feedback on some of the approaches I am trying, and if anyone has any other ideas or better methods...

Option 1: When a nested value object is required, create a custom type. In many cases, once I get to nested objects, they are wrapping a single primitive.

class CountryNameType(TypeDecorator):
 impl = String
 def process_bind_param(self, country_name_object, dialect):
 if country_name_object is not None:
 value = country_name_object.value
 return value
 def process_result_value(self, value, dialect):
 if value is not None:
 country_name_object = CountryName(value)
 return country_object

I was even thinking of creating generic types. Where a single type can be used for any VO with a string, numeric etc...:

SingleStringValueObjectType
SingleNumericValueObjectType

where they would have an init method that takes the class as an arg:

__init__(self, class_of_value_object)
super...

then pass in the class of the Type being represented. This could then be used in the two process methods to dynamically check it is the correct class type and create the new object from the passed in class. You could even pass in the generic type (String, Numeric) as an arg and use load_dialect_impl() to create a single generic type for any single primitive value object...

then

class Customer(Base)
first_name = Column(String)
last_name = Column(String)
country_name = Column(CountryNameType)
region = Column(RegionType)

or

country_name = Column(SingleStringValueObjectType(CountryName))
region = Column(SingleStringValueObjectType(Region))

with

customer_name = composite(CustomerName, first_name, last_name)
country = composite(Country, country_name, region)

Option 2:

Use @hybrid_property in the class to map the value objects.

ie.

@hybrid_property
def country(self):
 country_name = CountryName(self.db_country_name)
 region = Region(self.db_country_region)
 return Country(country_name, region)
@country.setter
def country(self, country):
 self.db_country_name = country.name.value
 self.db_country_region = country.region.value
 self._country = country

when db_country_name and db_country_region are defined as columns in the table.

class Customer(Base)

db_country_name = Column(String)
db_country_region = Column(String)
...

but are only ever set and retrieved through the hybrid property.

Custom Types feels like cleaner. But I'd be interested if anyone has any experience with any other (better) solutions?

Answer 1

Michael Bayer shared some very useful code with me on a thread on Bitbucket.

He uses a custom comparator directly with the composite, to illustrate how you can achieve a 2-level composite / comparator.

You can see the discussion and code here:

https://bitbucket.org/zzzeek/sqlalchemy/issues/4168/nested-composite-column-types

For simple value objects I created a Custom Type that seems to work as well:

class UnaryValueObjectType(TypeDecorator):
 impl = Numeric
 def __init__(self, class_of_value_object, type):
 self.class_of_value_object = class_of_value_object
 self.type = type
 super(UnaryValueObjectType, self).__init__()
 def load_dialect_impl(self, dialect):
 return dialect.type_descriptor(self.type)
 def process_bind_param(self, value_object, dialect):
 if isinstance(value_object, self.class_of_value_object) and value_object is not None:
 value = value_object.value
 return value
 def process_result_value(self, value, dialect):
 if value is not None:
 value_object = self.class_of_value_object(value)
 return value_object

Then

Column("country_name", UnaryValueObjectType(CountryName, String))

Hopefully both of these approaches can be useful to someone. They both meet my use cases (1 short, quick solution, another more powerful, but longer one).

• Steven: This has not been officially documented in the SQLAlchemy support docs: docs.sqlalchemy.org/en/13/orm/…
• Win Myo Htet: I cannot view that bitbucket thread even after creating an account on bitbucket. A bit disappointing for such a serious wall gardening mentality.