Skip to content

External Models

External Models provide read-only access to existing database tables without AMSDAL schema management. They map directly to tables in external databases — no migrations, no metadata tracking, no versioning.

Defining an External Model

from amsdal_models.classes.external_model import ExternalModel

class ExternalUser(ExternalModel):
    __table_name__ = 'users'
    __connection__ = 'external_users_db'
    __primary_key__ = ['id']

    id: int
    username: str
    email: str
    active: bool
Attribute Type Description
__table_name__ str Name of the table in the external database
__connection__ str Connection name (from your config)
__primary_key__ list[str] Primary key field(s). Defaults to ['id']

Querying

External Models use ExternalManager which provides a familiar QuerySet-like API. All queries end with .execute() (sync) or .aexecute() (async).

Basic queries

# All records
users = ExternalUser.objects.all().execute()

# Filter
active_users = ExternalUser.objects.filter(active=True).execute()

# Get single record (raises if not found or multiple)
user = ExternalUser.objects.get(id=1).execute()

# First record or None
user = ExternalUser.objects.first(active=True).execute()

# Count
total = ExternalUser.objects.count().execute()

# Exists
has_users = ExternalUser.objects.exists().execute()

Chaining

users = (
    ExternalUser.objects
    .filter(active=True)
    .filter(age__gte=18)
    .order_by('-created_at')
    .limit(10)
    .offset(20)
    .execute()
)

Field lookups

Lookup Example SQL equivalent
exact (default) filter(username='alice') = 'alice'
__gt filter(age__gt=28) > 28
__gte filter(age__gte=28) >= 28
__lt filter(age__lt=30) < 30
__lte filter(age__lte=30) <= 30
__in filter(username__in=['alice', 'bob']) IN (...)
__isnull filter(age__isnull=False) IS NOT NULL
__contains filter(email__contains='alice') LIKE '%alice%'
__icontains filter(email__icontains='alice') case-insensitive LIKE
__startswith filter(username__startswith='a') LIKE 'a%'
__endswith filter(username__endswith='e') LIKE '%e'

Async usage

users = await ExternalUser.objects.filter(active=True).aexecute()
user = await ExternalUser.objects.get(id=1).aexecute()

ExternalModelGenerator

Generate ExternalModel classes at runtime from an existing database schema.

from amsdal.services.external_model_generator import ExternalModelGenerator

generator = ExternalModelGenerator()

# Generate a single model
User = generator.generate_model('external_db', 'users')
users = User.objects.filter(active=True).execute()

# Generate all models for a connection
models = generator.generate_models_for_connection('external_db')
User = models['User']
Post = models['Post']

# Generate specific tables only
models = generator.generate_models_for_connection(
    'external_db',
    table_names=['users', 'posts'],
)

The generator automatically maps SQL types to Python types:

SQL Type Python Type
INTEGER, BIGINT int
TEXT, VARCHAR, CHAR str
REAL, FLOAT, DOUBLE float
BOOLEAN, BOOL bool
BLOB, BINARY bytes
DATE, DATETIME, TIMESTAMP str

ExternalDatabaseReader

For raw SQL queries against external databases:

from amsdal.services.external_connections import ExternalDatabaseReader

reader = ExternalDatabaseReader('external_users_db')

# Fetch all rows
users = reader.fetch_all('SELECT * FROM users WHERE active = ?', (1,))

# Fetch as dictionaries
user_dicts = reader.fetch_all_as_dicts('SELECT * FROM users LIMIT 10')

# Fetch one
user = reader.fetch_one('SELECT * FROM users WHERE id = ?', (42,))

# Introspection
tables = reader.get_table_names()
schema = reader.get_table_schema('users')

# Aggregates
total = reader.count('users')
exists = reader.exists('users', 'id = ?', (42,))

Warning

Always use parameterized queries (? placeholders) to prevent SQL injection.


CLI: Generate External Models

Generate model files from an external database:

amsdal generate external-models <connection_name> [OPTIONS]

Aliases: ext-models, em

Option Short Description
--output -o Output directory (default: models/external/)
--table -t Specific table(s) to generate (repeatable). All tables if omitted.
--format -f python (default) or json
--schema -s Database schema (e.g. public for PostgreSQL)

Examples:

# Generate all tables
amsdal generate external-models my_external_db

# Generate specific tables
amsdal generate external-models my_external_db -t users -t posts

# Output as JSON schema
amsdal generate external-models my_external_db --format json

# Custom output directory
amsdal generate external-models my_external_db -o src/external_models

Generated Python output:

from amsdal_models.classes.external_model import ExternalModel


class Users(ExternalModel):
    __table_name__ = 'users'
    __connection__ = 'external_db'

    id: int
    username: str
    email: str
    age: int | None = None
    active: bool = True

Configuration

External connections are defined in your application config:

from amsdal_utils.config.data_models.amsdal_config import AmsdalConfig
from amsdal_utils.config.data_models.connection_config import ConnectionConfig, ConnectionType
from amsdal_utils.config.data_models.resources_config import ResourcesConfig

config = AmsdalConfig(
    application_name='my_app',
    connections={
        'external_db': ConnectionConfig(
            name='external_db',
            backend='amsdal_data.connections.external.read_only_sqlite.ReadOnlySqliteConnection',
            credentials={'db_path': '/path/to/external.db'},
            connection_type=ConnectionType.EXTERNAL_SERVICE,
            is_managed=True,
        ),
    },
    resources_config=ResourcesConfig(
        external_services={'users_db': 'external_db'},
    ),
)

Note

External Models are read-only. They do not support save(), delete(), or migrations.