I have 2 different apps "Steam" and "Gasoline", this two classes has their unique operations but they also should have shared operations from a monorepo.
BaseOperation:
"TURBINE"
SteamOperations:
"TRAIN"
GasolineOperations:
"CAR"
So at the end Steam and Gasoline should extend from BaseOperations and have their unique Operation with shared operation "TURBINE".
Current Problems
I came out with below approach to combine all the operations.
class BaseOperation(enum.Enum):
TURBINE= "TURBINE"
class SteamOperation(enum.Enum):
TRAIN = "TRAIN"
class GasolineOperation(enum.Enum):
CAR= "CAR"
ALL_OPERATIONS = (
[e.value for e in BaseOperation] +
[e.value for e in SteamOperation] +
[e.value for e in GasolineOperation]
)
class OPEntry(declarative_base()):
operation = Column(Enum(*ALL_OPERATIONS), nullable=False)
But this approch has few issues. For example in SteamRouter I'm trying to pull both "TURBINE" and "TRAIN" but it is not extending and only "TRAIN" is there. Same goes for GasolineRouter
@router.post("/op-entry")
def add_op_entry(
operation: SteamOperations,
...)
Creating some CombinedOperations class is also not usefull because after any update in BaseOperations I would have to update all the other Services CombinedOperation classes.
How can I build my core model to include both unique and shared operations and have endpoint to manage both operations? What is the best way to use SQLAlchemy Enum with Dynamic Expansion?
Instead of using separate Enum
classes for SteamOperation
and GasolineOperation
, create a meta class that dynamically merges enums while keeping them manageable in SQL Alchemy.
class BaseOperation(enum.Enum):
TURBINE = "TURBINE"
class SteamOperation(enum.Enum):
TRAIN = "TRAIN"
class GasolineOperation(enum.Enum):
CAR = "CAR"
def merge_enums(name, base_enum, *additional_enums):
"""Dynamically merge multiple Enums into a single class"""
merged = {e.name: e.value for e in base_enum} # Start with base
for enum_class in additional_enums:
merged.update({e.name: e.value for e in enum_class}) # Add more
return enum.Enum(name, merged)
Now you can generate dynamic enums:
SteamOperations = merge_enums("SteamOperations", BaseOperation, SteamOperation)
GasolineOperations = merge_enums("GasolineOperations", BaseOperation, GasolineOperation)
class OPEntry(declarative_base()):
operation = Column(String, nullable=False) # Store as string for future expansions
Instead of manually listing all operations, use the dynamically created enum.
class OpEntryRequest(BaseModel):
operation: str # Store operations as string for flexibility
@router.post("/op-entry")
def add_op_entry(request: OpEntryRequest):
valid_operations = {e.value for e in SteamOperations} # Includes shared and unique
if request.operation not in valid_operations:
raise HTTPException(status_code=400, detail="Invalid operation type")
entry = OPEntry(operation=request.operation)
return {"message": f"Operation {entry.operation} added successfully"}