这是按照您的要求进行操作的基本方法
class Sample_Table(Base):
__tablename__ = 'Sample_Table'
__table_args__ = {'sqlite_autoincrement': True}
id = Column(Integer, primary_key=True, nullable=False)
col1 = Column(Integer)
col2 = Column(Integer)
def __init__(self, **kwargs):
for k,v in kwargs.items():
col_type = str(self.__table__.c[k].type)
try:
if str(type(v).__name__) in col_type.lower():
setattr(self, k, v)
else:
raise Exception("BAD COLUMN TYPE FOR COL " + k)
except ValueError as e:
print e.message
如果您尝试使用上述方法插入与您指定的列类型不同的记录,则会引发错误,即不会执行插入和回滚。
为了证明这是可行的,请尝试以下完整的代码:
from sqlalchemy import Column, Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
Base = declarative_base()
class Sample_Table(Base):
__tablename__ = 'Sample_Table'
__table_args__ = {'sqlite_autoincrement': True}
id = Column(Integer, primary_key=True, nullable=False)
col1 = Column(Integer)
col2 = Column(Integer)
def __init__(self, **kwargs):
for k,v in kwargs.items():
col_type = str(self.__table__.c[k].type)
try:
if str(type(v).__name__) in col_type.lower():
setattr(self, k, v)
else:
raise Exception("BAD COLUMN TYPE FOR COL " + k)
except ValueError as e:
print e.message
engine = create_engine('sqlite:///')
session = sessionmaker()
session.configure(bind=engine)
s = session()
Base.metadata.create_all(engine)
data = {"col1" : 1, "col2" : 2}
record = Sample_Table(**data)
s.add(record) #works
s.commit()
data = {"col1" : 1, "col2" : "2"}
record = Sample_Table(**data)
s.add(record) #doesn't work!
s.commit()
s.close()
(尽管我使用了 SQLite,但它也适用于类似的 MySQL 数据库。)