CODEWITHSUNDEEP
pythondatabasesqliteoopsqlalchemy
jfaccioni
jfaccioni

Reputation: 7529

Completely restart/reload declarative class with dynamic functionality in SQLAlchemy

I am using SQLAlchemy + SQLite3 for creating multiple databases based on user input. When initializing a new database, the user defines any number of arbitrary features and their types. I wrote a DBManager class to serve as an interface between user input and database creation/access.

Dynamically "injecting" these arbitrary features in the declarative model (the Features class) is working as expected. The problem I have is when the user wants to create a second/different database: I can't figure out how to completely "clear" or "refresh" the model or the declarative_base so that the user is able to create a new database (with possibly different features).

Below is a minimal reproducible example of my situation:

src.__init__.py:

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Session = sessionmaker()
Base = declarative_base()

src.features.py

from sqlalchemy import Column, ForeignKey, Integer
from sqlalchemy.orm import relationship

from src import Base

class Features(Base):
    __tablename__ = "features"
    features_id = Column(Integer, primary_key=True)

    @classmethod
    def add_feature(cls, feature_name, feature_type):
        setattr(cls, feature_name, Column(feature_type))

src.db_manager.py:

from typing import Optional, Dict

from sqlalchemy import create_engine

from src import Base, Session
from src.features import Features


class DBManager:
    def __init__(self, path: str, features: Optional[Dict] = None) -> None:
        self.engine = create_engine(f'sqlite:///{path}')
        Session.configure(bind=self.engine)
        self.session = Session()
        self.features = features
        if self.features:  # user passed in some arbitrary features
            self.bind_features_to_features_table()
        Base.metadata.create_all(bind=self.engine)

    def bind_features_to_features_table(self):
        for feature_name, feature_type in self.features.items():
            Features.add_feature(feature_name=feature_name, feature_type=feature_type)

I'd like to be able to do something like this:

from sqlalchemy import String, Float, Integer
from src.db_manager import DBManager

# User wants to create a database with these features
features = {
    'name': String,
    'height': Float,
}
db_manager = DBManager(path='my_database.db', features=features)

# ... User does some stuff with database here ...

# Now the user wants to create another database with these features
other_features = {
    'age': Integer,
    'weight': Float,
    'city_of_residence': String,
    'name': String,
}
db_manager = DBManager(path='another_database.db', features=other_features)

After executing the last line, I'm met with: InvalidRequestError: Implicitly combining column features.name with column features.name under attribute 'name'. Please configure one or more attributes for these same-named columns explicitly. The error wouldn't occur if the feature name did not appear on both databases, but then the feature height would be brought over to the second database, which is not desired.

Things I tried but didn't work:

This program will be running inside a GUI, so I can't really afford to exit/restart the script in order to connect to the second database. The user should be able to load/create different databases without having to close the program.

I understand that the error stems from the fact that the underlying Base object has not been "refreshed" and is still keeping track of the features created in my first DBManager instance. However I do not know how to fix this. What's worse, any attempt to overwrite/reload a new Base object will need to be applied to all modules that imported that object from __init__.py, which sounds tricky. Does anyone have a solution for this?

Upvotes: 1

Views: 442

Answers (1)

jfaccioni
jfaccioni

Reputation: 7529

My solution was to define the Features declarative class inside a function, get_features, that takes a Base (declarative base) instance as an argument. The function returns the Features class object, so that every call essentially creates a new Features class as a whole.

The class DBManager is then responsible for calling that function, and Features becomes a instance attribute of DBManager. Creating a new instance of DBManager means creating an entire new class based on Features, to which I can then add any arbitrary features I'd like.

The code looks something like this:

def get_features(declarative_base):
    class Features(declarative_base):
        __tablename__ = "features"
        features_id = Column(Integer, primary_key=True)

        @classmethod
        def add_feature(cls, feature_name, feature_type):
            setattr(cls, feature_name, Column(feature_type))

    return Features


class DBManager:
    def __init__(self, path, features):
        self.engine = create_engine(f'sqlite:///{path}')
        Session.configure(bind=self.engine)
        self.session = Session()
        base = declarative_base()
        self.features_table = get_features(base=base)
        if self.features:  # user passed in some arbitrary features
            self.bind_features_to_features_table()
        Base.metadata.create_all(bind=self.engine)

    def bind_features_to_features_table(self):
        for feature_name, feature_type in self.features.items():
            self.features_table.add_feature(feature_name=feature_name, feature_type=feature_type)

It definitely feels a bit convoluted, and I have no idea if there are any caveats I'm not aware of, but as far as I can tell this approach solved my problem.

Upvotes: 1

Related Questions