No history yet

Advanced Type Customization

Creating Custom Types

SQLAlchemy's built-in types cover most common use cases, but sometimes your application's domain requires something more specific. You might need to store data in a special format, validate it according to business rules, or work with a custom Python object that doesn't map directly to a standard database type. This is where TypeDecorator comes in.

TypeDecorator acts as a wrapper around an existing SQLAlchemy type, allowing you to intercept and transform data as it moves between your Python application and the database. It lets you create custom, reusable types that enforce domain-specific logic, making your code cleaner and your data more reliable.

Classes enable us to create our own custom data types that model real-world entities.

The core idea is to define two key behaviors: how to process a Python object before sending it to the database, and how to process a database value when it's loaded back into a Python object. Let's build a practical example: a type that automatically compresses and decompresses string data.

Building a CompressedString Type

Imagine you need to store large blocks of text efficiently. Database-level compression is an option, but handling it at the application level gives you more control. We'll create a CompressedString type that uses zlib to compress data before storage and decompress it upon retrieval. This process is transparent to the rest of the application, which can continue to work with regular Python strings.

Our custom type will wrap SQLAlchemy's LargeBinary type, since compressed data is binary.

import zlib
from sqlalchemy.types import TypeDecorator, LargeBinary

class CompressedString(TypeDecorator):
    """Saves compressed string data to the database."""
    impl = LargeBinary
    cache_ok = True # Indicates the type is safe to cache

    def process_bind_param(self, value, dialect):
        # Called when sending data to the database
        if value is not None:
            return zlib.compress(value.encode('utf-8'))
        return None

    def process_result_value(self, value, dialect):
        # Called when retrieving data from the database
        if value is not None:
            return zlib.decompress(value).decode('utf-8')
        return None

Let's break down the key methods.

process_bind_param(self, value, dialect) This method intercepts the value right before it's sent to the database. Here, value is the Python string you assigned to your model's attribute. We check if it's not None, encode it to bytes, compress it, and return the result. The database driver then receives these compressed bytes.

process_result_value(self, value, dialect) This method does the reverse. It receives the raw binary data from the database. We check for None, decompress the bytes, decode them back into a UTF-8 string, and return it. Your application code receives a clean, standard Python string, unaware of the compression that occurred.

Notice the careful handling of None in both methods. It ensures that NULL values in the database are correctly represented as None in your Python objects and vice-versa. The impl attribute tells SQLAlchemy which underlying database type to use when creating columns. In this case, we use LargeBinary.

Advanced: Dialect-Specific Logic

Sometimes, a custom type needs to behave differently depending on the database backend. For example, some databases have a native JSON type, while others might store JSON as plain text. The load_dialect_impl method allows you to return a different underlying type implementation based on the dialect in use.

Let's create a Url type. Our goal is to work with Python's pathlib.PurePosixPath objects in our code, but store them as simple strings in the database. For PostgreSQL, we might want to use a specific TEXT type, but for others, a generic String is fine.

from pathlib import PurePosixPath
from sqlalchemy.types import TypeDecorator, String, TEXT

class Url(TypeDecorator):
    """A custom type for storing and retrieving URL paths."""

    impl = String
    cache_ok = True

    def load_dialect_impl(self, dialect):
        # Return a specific type for a specific dialect
        if dialect.name == 'postgresql':
            return dialect.type_descriptor(TEXT())
        else:
            return dialect.type_descriptor(String(255))

    def process_bind_param(self, value, dialect):
        # Convert PurePosixPath to string for storage
        if value is not None:
            return str(value)
        return None

    def process_result_value(self, value, dialect):
        # Convert string from DB back to PurePosixPath
        if value is not None:
            return PurePosixPath(value)
        return None

Here, load_dialect_impl checks the dialect.name. If the backend is PostgreSQL, it returns a TEXT type. For all other databases, it returns a String with a length of 255. This allows you to fine-tune your column definitions for different environments without changing your model code.

The process_bind_param and process_result_value methods handle the conversion between the PurePosixPath object and its string representation. By abstracting this logic into a custom type, you maintain a clean separation of concerns and ensure data consistency.

Time to check your understanding of creating custom types with TypeDecorator.

Quiz Questions 1/5

What is the primary purpose of SQLAlchemy's TypeDecorator?

Quiz Questions 2/5

Which method in a TypeDecorator is responsible for converting a value from the database into a Python object?

Using TypeDecorator is a powerful technique for extending SQLAlchemy to perfectly match your application's data model. It bridges the gap between high-level Python objects and low-level database storage, keeping your application logic clean and your data integrity strong.