pythonsqlalchemyenumspydanticalembic

SQLAlchemy Enum: Combine Shared and Service-Specific Operations Dynamically


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?


Solution

  • 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.

    1. Define base and unique operations separately

    class BaseOperation(enum.Enum):
        TURBINE = "TURBINE"
    
    class SteamOperation(enum.Enum):
        TRAIN = "TRAIN"
    
    class GasolineOperation(enum.Enum):
        CAR = "CAR"
    

    2. Create a dynamic merged enum

    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)
    

    3. Use in SQLAlchemy model

    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.

    4. API

    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"}