Hi All,
I am using 0.7.10, sqlite memory and decl_enum.py from
http://techspot.zzzeek.org/2011/01/14/the-enum-recipe/
for testing.
Scenario 1 - test.py
I am running it 8 times, 7 out of 8 it works and I get the following output:
0.7.10
1 10 Action Main /gvv
2 20 Action Main /gvv1
3 30 Menu Main Sys
4 10 Action Test test
1 out of 8 I get the following error:
sqlalchemy.exc.IntegrityError: (IntegrityError) PRIMARY KEY must be unique
u'INSERT INTO "MenuItem" ("Id", "TypeOfMenuItem", "ItemMenu_Id", "Line",
"Desc", "CreateTimeStamp", "CreateOpId_Id", "ModifiedTimeStamp",
"ModifiedOpId_Id") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)' ((1, 'Action', 1,
10, 'Action 1', '2013-04-01 23:57:56.664654', 1, None, None), (2, 'Action',
1, 20, 'Action 2', '2013-04-01 23:57:56.664667', 1, None, None), (1,
'Menu', 1, 30, 'Menu', '2013-04-01 23:57:56.664674', 1, None, None))
Scenario 2 - test1.py
I have included table ProductClass just for noise. In scenario 1 it works
Adding index=True to ProductClass.Class:
Class = Column(String(16), index=True, nullable=False)
always gives me the following error:
sqlalchemy.exc.IntegrityError: (IntegrityError) PRIMARY KEY must be unique
u'INSERT INTO "MenuItem" ("Id", "TypeOfMenuItem", "ItemMenu_Id", "Line",
"Desc", "CreateTimeStamp", "CreateOpId_Id", "ModifiedTimeStamp",
"ModifiedOpId_Id") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)' (1, 'Menu', 1, 30,
'Menu', '2013-04-02 00:05:20.569187', 1, None, None)
What am I doing wrong?
Thank you in advance for your help.
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to sqlalchemy+unsubscr...@googlegroups.com.
To post to this group, send email to sqlalchemy@googlegroups.com.
Visit this group at http://groups.google.com/group/sqlalchemy?hl=en.
For more options, visit https://groups.google.com/groups/opt_out.
from sqlalchemy.types import SchemaType, TypeDecorator, Enum
from sqlalchemy import __version__
import re
if __version__ < '0.6.5':
raise NotImplementedError("Version 0.6.5 or higher of SQLAlchemy is required.")
class EnumSymbol(object):
"""Define a fixed symbol tied to a parent class."""
def __init__(self, cls_, name, value, description):
self.cls_ = cls_
self.name = name
self.value = value
self.description = description
def __reduce__(self):
"""Allow unpickling to return the symbol
linked to the DeclEnum class."""
return getattr, (self.cls_, self.name)
def __iter__(self):
return iter([self.value, self.description])
def __repr__(self):
return "<%s>" % self.name
class EnumMeta(type):
"""Generate new DeclEnum classes."""
def __init__(cls, classname, bases, dict_):
cls._reg = reg = cls._reg.copy()
for k, v in dict_.items():
if isinstance(v, tuple):
sym = reg[v[0]] = EnumSymbol(cls, k, *v)
setattr(cls, k, sym)
return type.__init__(cls, classname, bases, dict_)
def __iter__(cls):
return iter(cls._reg.values())
class DeclEnum(object):
"""Declarative enumeration."""
__metaclass__ = EnumMeta
_reg = {}
@classmethod
def from_string(cls, value):
try:
return cls._reg[value]
except KeyError:
raise ValueError(
"Invalid value for %r: %r" %
(cls.__name__, value)
)
@classmethod
def values(cls):
return cls._reg.keys()
@classmethod
def db_type(cls):
return DeclEnumType(cls)
class DeclEnumType(SchemaType, TypeDecorator):
def __init__(self, enum):
self.enum = enum
self.impl = Enum(
*enum.values(),
name="ck%s" % re.sub(
'([A-Z])',
lambda m:"_" + m.group(1).lower(),
enum.__name__)
)
def _set_table(self, table, column):
self.impl._set_table(table, column)
def copy(self):
return DeclEnumType(self.enum)
def process_bind_param(self, value, dialect):
if value is None:
return None
return value.value
def process_result_value(self, value, dialect):
if value is None:
return None
return self.enum.from_string(value.strip())
if __name__ == '__main__':
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import Session
Base = declarative_base()
class EmployeeType(DeclEnum):
part_time = "P", "Part Time"
full_time = "F", "Full Time"
contractor = "C", "Contractor"
class Employee(Base):
__tablename__ = 'employee'
id = Column(Integer, primary_key=True)
name = Column(String(60), nullable=False)
type = Column(EmployeeType.db_type())
def __repr__(self):
return "Employee(%r, %r)" % (self.name, self.type)
e = create_engine('sqlite://', echo=True)
Base.metadata.create_all(e)
sess = Session(e)
sess.add_all([
Employee(name='e1', type=EmployeeType.full_time),
Employee(name='e2', type=EmployeeType.full_time),
Employee(name='e3', type=EmployeeType.part_time),
Employee(name='e4', type=EmployeeType.contractor),
Employee(name='e5', type=EmployeeType.contractor),
])
sess.commit()
print sess.query(Employee).filter_by(type=EmployeeType.contractor).all()from datetime import datetime
from sqlalchemy import Column, ForeignKey, String, Integer, PickleType, DateTime, Index, create_engine
from sqlalchemy import __version__
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, backref, Session
from decl_enum import DeclEnum
Base = declarative_base()
def GetCurrentUserOID():
return 1
class User(Base):
__tablename__ = 'User'
Id = Column(Integer, primary_key=True)
Name = Column(String(64))
CreateTimeStamp = Column(DateTime, default=datetime.now)
# ManyToOne
CreateOpId_Id = Column(Integer,ForeignKey('User.Id', use_alter=True, name="FK_User_CreateOp_Id", ondelete="SET NULL"), default=GetCurrentUserOID)
CreateOpId = relationship('User', primaryjoin='User.Id == User.CreateOpId_Id', remote_side="User.Id") # remote_side for self referencing User->User
ModifiedTimeStamp = Column(DateTime, onupdate=datetime.now)
# ManyToOne
ModifiedOpId_Id = Column(Integer,ForeignKey('User.Id', use_alter=True, name="FK_User_ModifiedOp_Id", ondelete="SET NULL"), onupdate=GetCurrentUserOID)
ModifiedOpId = relationship('User', primaryjoin='User.Id == User.ModifiedOpId_Id', remote_side="User.Id")
class Company(Base):
__tablename__ = 'Company'
Id = Column(Integer, primary_key=True)
Name = Column(String(64))
CreateTimeStamp = Column(DateTime, default=datetime.now)
# ManyToOne
CreateOpId_Id = Column(Integer,ForeignKey('User.Id', use_alter=True, name="FK_Company_CreateOp_Id", ondelete="SET NULL"), default=GetCurrentUserOID)
CreateOpId = relationship('User', primaryjoin='User.Id == Company.CreateOpId_Id', remote_side="User.Id") # remote_side for self referencing User->User
ModifiedTimeStamp = Column(DateTime, onupdate=datetime.now)
# ManyToOne
ModifiedOpId_Id = Column(Integer,ForeignKey('User.Id', use_alter=True, name="FK_Company_ModifiedOp_Id", ondelete="SET NULL"), onupdate=GetCurrentUserOID)
ModifiedOpId = relationship('User', primaryjoin='User.Id == Company.ModifiedOpId_Id', remote_side="User.Id")
class ProductClass(Base):
__tablename__ = 'ProductClass'
Id = Column(Integer, primary_key=True)
# Many2One
Company_Id = Column(Integer, ForeignKey("Company.Id", use_alter=True, name="FK_ProductClass_Company_Id", ondelete="SET NULL"))
Company = relationship("Company")
Class = Column(String(16), nullable=False)
CreateTimeStamp = Column(DateTime, default=datetime.now)
# ManyToOne
CreateOpId_Id = Column(Integer,ForeignKey('User.Id', use_alter=True, name="FK_ProductClass_CreateOp_Id", ondelete="SET NULL"), default=GetCurrentUserOID)
CreateOpId = relationship('User', primaryjoin='User.Id == ProductClass.CreateOpId_Id')
ModifiedTimeStamp = Column(DateTime, onupdate=datetime.now)
# ManyToOne
ModifiedOpId_Id = Column(Integer,ForeignKey('User.Id', use_alter=True, name="FK_ProductClass_ModifiedOp_Id", ondelete="SET NULL"), onupdate=GetCurrentUserOID)
ModifiedOpId = relationship('User', primaryjoin='User.Id == ProductClass.ModifiedOpId_Id')
Index("ProductClass_Index1", ProductClass.Company_Id, ProductClass.Class, unique=True)
class Menu(Base):
__tablename__ = 'Menu'
Id = Column(Integer, primary_key=True)
MenuId = Column(String(16), index=True, unique=True, nullable=False)
Name = Column(String(64), nullable=False)
# OneToMany side of MenuItem ManyToOne - backref MenuItem.ItemMenu
Items = relationship("MenuItem", uselist=True,
order_by="MenuItem.Line", cascade="delete", backref="ItemMenu")
CreateTimeStamp = Column(DateTime, default=datetime.now)
# ManyToOne
CreateOpId_Id = Column(Integer,ForeignKey('User.Id', use_alter=True, name="FK_Menu_CreateOp_Id", ondelete="SET NULL"), default=GetCurrentUserOID)
CreateOpId = relationship('User', primaryjoin='User.Id == Menu.CreateOpId_Id') # remote_side for self referencing User->User
ModifiedTimeStamp = Column(DateTime, onupdate=datetime.now)
# ManyToOne
ModifiedOpId_Id = Column(Integer,ForeignKey('User.Id', use_alter=True, name="FK_Menu_ModifiedOp_Id", ondelete="SET NULL"), onupdate=GetCurrentUserOID)
ModifiedOpId = relationship('User', primaryjoin='User.Id == Menu.ModifiedOpId_Id')
class MenuItemType(DeclEnum):
# name = value, description
action = "Action", "Action Item"
menu = "Menu", "Menu Item"
class MenuItem(Base):
__tablename__ = 'MenuItem'
Id = Column(Integer, primary_key=True)
TypeOfMenuItem = Column(MenuItemType.db_type(), nullable=False)
# ManyToOne side of Menu OneTomany - backref MenuItem.ItemMenu
ItemMenu_Id = Column(Integer,ForeignKey('Menu.Id', use_alter=True, name="FK_MenuItem_ItemMenu_Id"))
Line = Column(Integer, nullable=False)
Desc = Column(String(64), nullable=False)
CreateTimeStamp = Column(DateTime, default=datetime.now)
# ManyToOne
CreateOpId_Id = Column(Integer,ForeignKey('User.Id', use_alter=True, name="FK_MenuItem_CreateOp_Id", ondelete="SET NULL"), default=GetCurrentUserOID)
CreateOpId = relationship('User', primaryjoin='User.Id == MenuItem.CreateOpId_Id')
ModifiedTimeStamp = Column(DateTime, onupdate=datetime.now)
# ManyToOne
ModifiedOpId_Id = Column(Integer,ForeignKey('User.Id', use_alter=True, name="FK_MenuItem_ModifiedOp_Id", ondelete="SET NULL"), onupdate=GetCurrentUserOID)
ModifiedOpId = relationship('User', primaryjoin='User.Id == MenuItem.ModifiedOpId_Id')
__mapper_args__ = {'polymorphic_on': TypeOfMenuItem,
'with_polymorphic': '*'
}
class MenuItemAction(MenuItem):
__tablename__ = 'MenuItemAction'
Id = Column(Integer, primary_key=True)
__mapper_args__ = {'polymorphic_identity': MenuItemType.action,
'inherit_condition': (Id == MenuItem.Id)
}
Action = Column(String(64), nullable=False)
Params = Column(PickleType)
class MenuItemMenu(MenuItem):
__tablename__ = 'MenuItemMenu'
Id = Column(Integer, primary_key=True)
__mapper_args__ = {'polymorphic_identity': MenuItemType.menu,
'inherit_condition': (Id == MenuItem.Id)
}
# ManyToOne
Menu_Id = Column(Integer, ForeignKey('Menu.Id', use_alter=True, name="FK_MenuItemMenu_Menu_Id", ondelete="SET NULL"))
Menu = relationship('Menu')
engine = create_engine('sqlite://', echo=False)
Base.metadata.create_all(engine)
print __version__
useListDO = []
def newRecord(tableName, data):
session = Session(engine)
menu = _newRecord(tableName, data["Menu"], session)
# assign MenuItems to Menu
for do in useListDO:
setattr(do, "ItemMenu", menu)
session.commit()
def _newRecord(table, fields, session):
if table == "Menu":
do = Menu()
else:
do = MenuItemAction()
for fieldName, value in fields.iteritems():
if fieldName == "Items":
for i in value:
for t, f in i.iteritems():
useListDO.append(_newRecord(t, f, session))
continue
if fieldName == "TypeOfMenuItem":
continue
setattr(do, fieldName, value)
session.add(do)
return do
if __name__ == '__main__':
session = Session(engine)
# populate tables
user = User(Name="gvv")
session.add(user)
mainmenu = Menu(MenuId="Main", Name="Main Menu")
session.add(mainmenu)
sysmenu = Menu(MenuId="Sys", Name="System Menu")
session.add(sysmenu)
item = MenuItemAction(ItemMenu=mainmenu, Line=10, Desc="Action 1", Action="/gvv")
session.add(item)
item = MenuItemAction(Line=20, Desc="Action 2", Action="/gvv1")
session.add(item)
item.ItemMenu = mainmenu
item = MenuItemMenu(ItemMenu=mainmenu, Line=30, Desc="Menu", Menu=sysmenu)
session.add(item)
session.commit()
#
data = {'Menu': {'MenuId': 'Test', 'Items': [{'MenuItemAction': {'Action': 'test', 'TypeOfMenuItem': 'Action', 'Params': '', 'Line': '10', 'Desc': 'Test'}}], 'Name': 'Test'}}
table = "Menu"
newRecord(table, data)
items = session.query(MenuItem).all()
for i in items:
if i.TypeOfMenuItem.value == "Action":
print i.Id, i.Line, i.TypeOfMenuItem.value, i.ItemMenu.MenuId, i.Action
else:
print i.Id, i.Line, i.TypeOfMenuItem.value, i.ItemMenu.MenuId, i.Menu.MenuId
from datetime import datetime
from sqlalchemy import Column, ForeignKey, String, Integer, PickleType, DateTime, Index, create_engine
from sqlalchemy import __version__
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, backref, Session
from decl_enum import DeclEnum
Base = declarative_base()
def GetCurrentUserOID():
return 1
class User(Base):
__tablename__ = 'User'
Id = Column(Integer, primary_key=True)
Name = Column(String(64))
CreateTimeStamp = Column(DateTime, default=datetime.now)
# ManyToOne
CreateOpId_Id = Column(Integer,ForeignKey('User.Id', use_alter=True, name="FK_User_CreateOp_Id", ondelete="SET NULL"), default=GetCurrentUserOID)
CreateOpId = relationship('User', primaryjoin='User.Id == User.CreateOpId_Id', remote_side="User.Id") # remote_side for self referencing User->User
ModifiedTimeStamp = Column(DateTime, onupdate=datetime.now)
# ManyToOne
ModifiedOpId_Id = Column(Integer,ForeignKey('User.Id', use_alter=True, name="FK_User_ModifiedOp_Id", ondelete="SET NULL"), onupdate=GetCurrentUserOID)
ModifiedOpId = relationship('User', primaryjoin='User.Id == User.ModifiedOpId_Id', remote_side="User.Id")
class Company(Base):
__tablename__ = 'Company'
Id = Column(Integer, primary_key=True)
Name = Column(String(64))
CreateTimeStamp = Column(DateTime, default=datetime.now)
# ManyToOne
CreateOpId_Id = Column(Integer,ForeignKey('User.Id', use_alter=True, name="FK_Company_CreateOp_Id", ondelete="SET NULL"), default=GetCurrentUserOID)
CreateOpId = relationship('User', primaryjoin='User.Id == Company.CreateOpId_Id', remote_side="User.Id") # remote_side for self referencing User->User
ModifiedTimeStamp = Column(DateTime, onupdate=datetime.now)
# ManyToOne
ModifiedOpId_Id = Column(Integer,ForeignKey('User.Id', use_alter=True, name="FK_Company_ModifiedOp_Id", ondelete="SET NULL"), onupdate=GetCurrentUserOID)
ModifiedOpId = relationship('User', primaryjoin='User.Id == Company.ModifiedOpId_Id', remote_side="User.Id")
class ProductClass(Base):
__tablename__ = 'ProductClass'
Id = Column(Integer, primary_key=True)
# Many2One
Company_Id = Column(Integer, ForeignKey("Company.Id", use_alter=True, name="FK_ProductClass_Company_Id", ondelete="SET NULL"))
Company = relationship("Company")
Class = Column(String(16), index=True, nullable=False)
CreateTimeStamp = Column(DateTime, default=datetime.now)
# ManyToOne
CreateOpId_Id = Column(Integer,ForeignKey('User.Id', use_alter=True, name="FK_ProductClass_CreateOp_Id", ondelete="SET NULL"), default=GetCurrentUserOID)
CreateOpId = relationship('User', primaryjoin='User.Id == ProductClass.CreateOpId_Id')
ModifiedTimeStamp = Column(DateTime, onupdate=datetime.now)
# ManyToOne
ModifiedOpId_Id = Column(Integer,ForeignKey('User.Id', use_alter=True, name="FK_ProductClass_ModifiedOp_Id", ondelete="SET NULL"), onupdate=GetCurrentUserOID)
ModifiedOpId = relationship('User', primaryjoin='User.Id == ProductClass.ModifiedOpId_Id')
Index("ProductClass_Index1", ProductClass.Company_Id, ProductClass.Class, unique=True)
class Menu(Base):
__tablename__ = 'Menu'
Id = Column(Integer, primary_key=True)
MenuId = Column(String(16), index=True, unique=True, nullable=False)
Name = Column(String(64), nullable=False)
# OneToMany side of MenuItem ManyToOne - backref MenuItem.ItemMenu
Items = relationship("MenuItem", uselist=True,
order_by="MenuItem.Line", cascade="delete", backref="ItemMenu")
CreateTimeStamp = Column(DateTime, default=datetime.now)
# ManyToOne
CreateOpId_Id = Column(Integer,ForeignKey('User.Id', use_alter=True, name="FK_Menu_CreateOp_Id", ondelete="SET NULL"), default=GetCurrentUserOID)
CreateOpId = relationship('User', primaryjoin='User.Id == Menu.CreateOpId_Id') # remote_side for self referencing User->User
ModifiedTimeStamp = Column(DateTime, onupdate=datetime.now)
# ManyToOne
ModifiedOpId_Id = Column(Integer,ForeignKey('User.Id', use_alter=True, name="FK_Menu_ModifiedOp_Id", ondelete="SET NULL"), onupdate=GetCurrentUserOID)
ModifiedOpId = relationship('User', primaryjoin='User.Id == Menu.ModifiedOpId_Id')
class MenuItemType(DeclEnum):
# name = value, description
action = "Action", "Action Item"
menu = "Menu", "Menu Item"
class MenuItem(Base):
__tablename__ = 'MenuItem'
Id = Column(Integer, primary_key=True)
TypeOfMenuItem = Column(MenuItemType.db_type(), nullable=False)
# ManyToOne side of Menu OneTomany - backref MenuItem.ItemMenu
ItemMenu_Id = Column(Integer,ForeignKey('Menu.Id', use_alter=True, name="FK_MenuItem_ItemMenu_Id"))
Line = Column(Integer, nullable=False)
Desc = Column(String(64), nullable=False)
CreateTimeStamp = Column(DateTime, default=datetime.now)
# ManyToOne
CreateOpId_Id = Column(Integer,ForeignKey('User.Id', use_alter=True, name="FK_MenuItem_CreateOp_Id", ondelete="SET NULL"), default=GetCurrentUserOID)
CreateOpId = relationship('User', primaryjoin='User.Id == MenuItem.CreateOpId_Id')
ModifiedTimeStamp = Column(DateTime, onupdate=datetime.now)
# ManyToOne
ModifiedOpId_Id = Column(Integer,ForeignKey('User.Id', use_alter=True, name="FK_MenuItem_ModifiedOp_Id", ondelete="SET NULL"), onupdate=GetCurrentUserOID)
ModifiedOpId = relationship('User', primaryjoin='User.Id == MenuItem.ModifiedOpId_Id')
__mapper_args__ = {'polymorphic_on': TypeOfMenuItem,
'with_polymorphic': '*'
}
class MenuItemAction(MenuItem):
__tablename__ = 'MenuItemAction'
Id = Column(Integer, primary_key=True)
__mapper_args__ = {'polymorphic_identity': MenuItemType.action,
'inherit_condition': (Id == MenuItem.Id)
}
Action = Column(String(64), nullable=False)
Params = Column(PickleType)
class MenuItemMenu(MenuItem):
__tablename__ = 'MenuItemMenu'
Id = Column(Integer, primary_key=True)
__mapper_args__ = {'polymorphic_identity': MenuItemType.menu,
'inherit_condition': (Id == MenuItem.Id)
}
# ManyToOne
Menu_Id = Column(Integer, ForeignKey('Menu.Id', use_alter=True, name="FK_MenuItemMenu_Menu_Id", ondelete="SET NULL"))
Menu = relationship('Menu')
engine = create_engine('sqlite://', echo=False)
Base.metadata.create_all(engine)
print __version__
useListDO = []
def newRecord(tableName, data):
session = Session(engine)
menu = _newRecord(tableName, data["Menu"], session)
# assign MenuItems to Menu
for do in useListDO:
setattr(do, "ItemMenu", menu)
session.commit()
def _newRecord(table, fields, session):
if table == "Menu":
do = Menu()
else:
do = MenuItemAction()
for fieldName, value in fields.iteritems():
if fieldName == "Items":
for i in value:
for t, f in i.iteritems():
useListDO.append(_newRecord(t, f, session))
continue
if fieldName == "TypeOfMenuItem":
continue
setattr(do, fieldName, value)
session.add(do)
return do
if __name__ == '__main__':
session = Session(engine)
# populate tables
user = User(Name="gvv")
session.add(user)
mainmenu = Menu(MenuId="Main", Name="Main Menu")
session.add(mainmenu)
sysmenu = Menu(MenuId="Sys", Name="System Menu")
session.add(sysmenu)
item = MenuItemAction(ItemMenu=mainmenu, Line=10, Desc="Action 1", Action="/gvv")
session.add(item)
item = MenuItemAction(ItemMenu=mainmenu, Line=20, Desc="Action 2", Action="/gvv1")
session.add(item)
item = MenuItemMenu(ItemMenu=mainmenu, Line=30, Desc="Menu", Menu=sysmenu)
session.add(item)
session.commit()
#
data = {'Menu': {'MenuId': 'Test', 'Items': [{'MenuItemAction': {'Action': 'test', 'TypeOfMenuItem': 'Action', 'Params': '', 'Line': '10', 'Desc': 'Test'}}], 'Name': 'Test'}}
table = "Menu"
newRecord(table, data)
items = session.query(MenuItem).all()
for i in items:
if i.TypeOfMenuItem.value == "Action":
print i.Id, i.Line, i.TypeOfMenuItem.value, i.ItemMenu.MenuId, i.Action
else:
print i.Id, i.Line, i.TypeOfMenuItem.value, i.ItemMenu.MenuId, i.Menu.MenuId
