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.