Skip to main content

Data Module Overview

info

The Data module provides data access, persistence, and query capabilities with support for multiple database backends.

Introduction​

The Data module abstracts data access and provides:

  • Data Drivers: Support for SQLAlchemy, MongoDB, and in-memory storage
  • Data Models: Pydantic-based models with validation
  • Data Schemas: SQLAlchemy schema definitions
  • Query Building: Type-safe query construction
  • Serialization: JSON encoding/decoding with custom encoders

Key Components​

DataModel​

DataModel is the base class for all data models:

from fluvius.data import DataModel, field

class User(DataModel):
    name: str = field()
    email: str = field()
    active: bool = field(initial=True)

DataDriver​

Data drivers provide database-specific implementations:

from fluvius.data import SqlaDriver

driver = SqlaDriver(connection_string='postgresql://...')

DataManager​

The Data Manager provides high-level data access:

from fluvius.data import DataAccessManager

manager = DataAccessManager(driver)

# Fetch entity
user = await manager.fetch('user', user_id)

# Find entities
users = await manager.find('user', active=True)

# Save entity
await manager.save('user', user_id, user_data)

DataSchema​

Schemas define database table structures:

from fluvius.data import SqlaDataSchema
from sqlalchemy import Column, String, Boolean

class UserSchema(SqlaDataSchema):
    __tablename__ = 'users'
    
    name = Column(String)
    email = Column(String)
    active = Column(Boolean, default=True)

Supported Backends​

SQLAlchemy (PostgreSQL, SQLite, MySQL)​

from fluvius.data import SqlaDriver

driver = SqlaDriver(
    connection_string='postgresql://user:pass@localhost/db'
)

MongoDB​

from fluvius.data.data_driver.mongodb import MongoDriver

driver = MongoDriver(
    connection_string='mongodb://localhost:27017/'
)

In-Memory​

from fluvius.data.data_driver.memory import MemoryDriver

driver = MemoryDriver()

Query Building​

Build type-safe queries:

from fluvius.data import BackendQuery

query = BackendQuery('user').filter(active=True).limit(10)
results = await manager.query(query)

Serialization​

Custom JSON encoding for complex types:

from fluvius.data import JSONEncoder, serialize_json

data = {'user_id': UUID_GENR(), 'timestamp': datetime.now()}
json_str = serialize_json(data, encoder=JSONEncoder)

Next Steps​