Reputation: 21914
I would like to use SQLAlchemy to add some test fixtures to the database. It should should get
or set
the value accordingly. I was trying to create a model method like: ProductsModel.ensure(name="Dettol")
that will ensure the data in the DB against the name
column, where Products
is defined like:
Products
class Products(Base):
__tablename__ = 'products'
idn = Column(Integer)
name = Column(String, primary_key=True)
ProductsModel.ensure
will take care of doing the unique-row check against the composite keys (name
in this case) and decide if it has to be get or set.
However, I can't modify the real Products
table definition which is defined (elsewhere) with id
as the primary key like this:
class Products(Base):
__tablename__ = 'products'
idn = Column(Integer, primary_key=True)
name = Column(String)
So, is it possible for me to inherit the table, and modify the primary keys for my test-fixture use case?
I tried something like this:
class ProductsUnique(Products):
__mapper_args__ = {
'polymorphic_identity':'productsunique',
'polymorphic_on':Products.idn # Hoping that I could override idn.
}
idn = Column(Integer)
name = Column(String, primary_key=True)
But it says:
sqlalchemy.exc.ArgumentError: Column 'idn' on class <class '__main__.ProductsUnique'> conflicts with existing column 'products.idn'
I believe, setting polymorphic_on
on the original table might resolve the issue; but the problem is that I can't do it there.
Is there a way to set this after the class has been defined? There seems to be an open issue similar to this in the SQLAlchemy repository.
PS: After reading a lot of posts, I think I am looking for something like Django's get_or_create; but I would like to more clear about the natural keys for my table. I am not sure if there are existing solutions to solve this problem.
Upvotes: 0
Views: 55
Reputation: 6301
Why do you need to modify the primary key in order to check whether a row exists? You can't use Session.merge
, but you could implement the check yourself:
class Product:
@classmethod
def get_or_create(cls, name):
query = session.query(Product).filter_by(name=name)
result = query.one_or_none()
if not result:
result = Product(name=name)
session.add(result)
session.commit()
# maybe load again here to prevent an unexpected one later
return result
If you're worried about race conditions:
Upvotes: 1