ObjDataModel is a generic CRUD layer that eliminates the need for per-domain-object subclasses. Instead of creating separate classes like ObjCreditApplication, ObjCreditContract, and ObjPaymentSchedule, all domain models are configured via YAML schema definitions and managed through a single ObjDataModel class.
Before: Each domain object required a dedicated class + YAML + hand-written SQL
ObjPerson.py + ObjPerson.yaml + SQL queriesObjCreditApplication.py + ObjCreditApplication.yaml + SQL queriesObjCreditContract.py + ObjCreditContract.yaml + SQL queriesread(), save(), _load_from_row(), SQL generationAfter: Schema-driven, no subclassing
app = ObjDataModel("credit_application", DB=db)
app.from_dict(request_data)
errors = app.validate()
if not errors:
app.save()
Schemas are defined in per-package models.yaml files under the models: section.
factory.report/
package.fullhouse/
models.yaml ← Define credit_application, credit_contract, payment_schedule
ObjReportCreditJourney.py
models:
credit_application:
table: data_credit_application
guid_field: ApplicationGuid
package_field: Package
active_field: Active
default_sort: SubmissionDate DESC
fields:
ApplicationGuid:
type: char(50)
primary_key: true
required: true
FirstName:
type: varchar(255)
required: true
MonthlyIncome:
type: decimal(10,2)
required: true
min: 5000
Status:
type: varchar(50)
default: pending
enum: [pending, approved, rejected]
Active:
type: char(1)
default: "Y"
LastUpdateDate:
type: datetime
auto_now: true
| Option | Type | Purpose | Example |
|---|---|---|---|
type |
string | SQL column type | varchar(255), decimal(10,2), datetime |
primary_key |
bool | Mark as PRIMARY KEY | primary_key: true |
required |
bool | Reject empty values | required: true |
default |
any | SQL DEFAULT value | default: pending |
auto_now |
bool | Always set to NOW() on save | auto_now: true |
enum |
list | Allowed values | enum: [pending, approved, rejected] |
max_length |
int | String max length | max_length: 50 |
min |
number | Numeric minimum | min: 5000 |
max |
number | Numeric maximum | max: 150000 |
from factory.core import ObjDataModel
# Create instance (loads schema, creates table)
app = ObjDataModel("credit_application", DB=self.DB)
# Populate from RPC request
app.from_dict({
"FirstName": "John",
"LastName": "Doe",
"Email": "john@example.com",
"MonthlyIncome": 25000,
"Status": "pending"
})
errors = app.validate()
if errors:
return {"status": "error", "errors": errors}
# ["FirstName is required", "MonthlyIncome must be >= 5000"]
success = app.save()
if success:
guid = app.get("ApplicationGuid")
return {"status": "ok", "guid": guid}
app2 = ObjDataModel("credit_application", DB=self.DB)
found = app2.read("20260411_1430_abc123_")
if found:
print(app2.get("FirstName")) # "John"
print(app2.to_dict()) # Full record as dict
pending = app.list(filters={"Status": "pending"})
approved = app.list(filters={"Status": "approved"}, active_only=True)
for record in pending:
print(record["FirstName"], record["MonthlyIncome"])
app.set("Status", "approved")
app.set("ApprovedLimit", 50000)
app.save()
app.delete("20260411_1430_abc123_")
# Sets Active='N' instead of hard delete
# Get a field
income = app.get("MonthlyIncome")
status = app.get("Status", default="pending")
# Set a field
app.set("FirstName", "Jane")
app.set("MonthlyIncome", 35000)
# Export all fields as dict
data = app.to_dict()
# {"FirstName": "Jane", "MonthlyIncome": 35000, ...}
# Import from dict
app.from_dict(data)
validate() returns a list of error strings (empty = valid):
errors = app.validate()
if errors:
for error in errors:
print(f" - {error}")
# Output:
# - FirstName is required
# - MonthlyIncome must be >= 5000
# - Status must be one of: pending, approved, rejected
Validation rules checked:
required: true → error if emptyenum: [...] → error if value not in listmax_length: N → error if string exceeds Nmin: N / max: N → error if numeric value out of rangeObjDataModel generates all SQL from the schema. No hand-written queries.
CREATE TABLE IF NOT EXISTS data_credit_application (
ApplicationGuid char(50) PRIMARY KEY NOT NULL,
FirstName varchar(255) NOT NULL,
MonthlyIncome decimal(10,2) NOT NULL,
Status varchar(50) DEFAULT 'pending',
Active char(1) DEFAULT 'Y',
LastUpdateDate datetime
)
INSERT INTO data_credit_application
(ApplicationGuid, FirstName, MonthlyIncome, Status, LastUpdateDate)
VALUES ('20260411_...', 'John', 25000, 'pending', NOW())
ON DUPLICATE KEY UPDATE
FirstName = 'John',
MonthlyIncome = 25000,
Status = 'pending',
LastUpdateDate = NOW()
SELECT * FROM data_credit_application
WHERE ApplicationGuid = '20260411_...'
SELECT * FROM data_credit_application
WHERE Package = 'fullhouse'
AND Active = 'Y'
AND Status = 'approved'
ORDER BY SubmissionDate DESC
Use ObjDataModel in RPC handlers to eliminate boilerplate:
class Report(ObjData.ObjData):
def RpcSubmitApplication(self, data):
"""Submit new credit application"""
app = ObjDataModel.ObjDataModel("credit_application", DB=self.DB)
app.from_dict(data)
errors = app.validate()
if errors:
return {"status": "error", "errors": errors}
app.save()
return {
"status": "ok",
"guid": app.get("ApplicationGuid"),
"application": app.to_dict()
}
def RpcGetApplication(self, data):
"""Retrieve application by GUID"""
app = ObjDataModel.ObjDataModel("credit_application", DB=self.DB)
if app.read(data.get("guid")):
return {"status": "ok", "application": app.to_dict()}
else:
return {"status": "error", "message": "Application not found"}
def RpcListApplications(self, data):
"""List applications by status"""
app = ObjDataModel.ObjDataModel("credit_application", DB=self.DB)
records = app.list(
filters={"Status": data.get("status", "pending")},
active_only=True
)
return {
"status": "ok",
"count": len(records),
"applications": records
}
Schemas are cached in memory with thread-safe locking. First instantiation of a schema name loads and caches the YAML; subsequent instantiations reuse the cached schema.
# First call: loads YAML, caches schema
app1 = ObjDataModel("credit_application", DB=db)
# Subsequent calls: reuse cached schema (fast)
app2 = ObjDataModel("credit_application", DB=db)
app3 = ObjDataModel("credit_application", DB=db)
✅ Schema-driven — No subclass code needed
✅ Auto table creation — CREATE TABLE generated from fields
✅ GUID generation — Uses get_uuid() for timestamp-prefixed IDs
✅ Validation — Required, enum, range, length checks
✅ Soft delete — Sets Active='N' instead of hard delete
✅ Package filtering — All queries automatically filter by package
✅ auto_now fields — LastUpdateDate always set to NOW() on save
✅ Thread-safe caching — Schema cache with RLock
✅ SQL parameterization — Uses escape_sql() for all values
✅ Flexible — Works with any domain model
ObjData.ObjData ← Base database class
↑
└─ ObjDataModel ← Generic schema-driven model
| Method | Signature | Purpose |
|---|---|---|
read(guid) |
read(guid: str) -> bool |
Load by GUID |
save() |
save() -> bool |
INSERT/UPDATE with validation |
delete(guid) |
delete(guid: str) -> bool |
Soft delete |
list(filters, active_only) |
list(...) -> List[Dict] |
Query with filters |
get(field, default) |
get(field: str) -> Any |
Get field value |
set(field, value) |
set(field: str, value: Any) |
Set field value |
from_dict(data) |
from_dict(data: Dict) -> None |
Populate from dict |
to_dict() |
to_dict() -> Dict |
Export all fields |
validate() |
validate() -> List[str] |
Validate and return errors |
| Method | Purpose |
|---|---|
_load_schema(schema_name) |
Load schema from YAML |
_ensure_table() |
CREATE TABLE if not exists |
_build_create_table_sql() |
Generate CREATE TABLE |
_build_select_sql(guid) |
Generate SELECT by GUID |
_build_upsert_sql() |
Generate INSERT...ON DUPLICATE |
_build_list_sql(filters) |
Generate SELECT with filters |
_load_from_row(row) |
Populate from database row |
Test suite: resource.test/objdatamodel_unit_test.py
cd ~/projects/axion
python3 resource.test/objdatamodel_unit_test.py
Tests schema loading, SQL generation, field access, and validation.
# Submit application
app = ObjDataModel("credit_application", DB=db)
app.from_dict({"FirstName": "John", ...})
app.save()
# Later: approve with limit
app.read(guid)
app.set("Status", "approved")
app.set("ApprovedLimit", 50000)
app.set("ApprovedTerm", 24)
app.save()
# Later: reject
app.set("Status", "rejected")
app.set("RejectionReason", "Income too low")
app.save()
# After approval, create contract
app = ObjDataModel("credit_application", DB=db)
app.read(app_guid)
contract = ObjDataModel("credit_contract", DB=db)
contract.set("ApplicationGuid", app.get("ApplicationGuid"))
contract.set("CreditAmount", app.get("ApprovedLimit"))
contract.set("PaymentTerm", app.get("ApprovedTerm"))
contract.save()
char(50) to fit get_uuid() outputlist() queriesauto_now fields are always updated to NOW() on saveVersion: 1.0
Author: vheyn@technocore.co.za
Status: Production-ready