请问,flask-sqlalchemy 怎么动态创建数据表?

发布网友发布时间：2022-04-08 01:42

共2个回答

懂视网时间：2022-04-08 06:03

class GoodsDesc(db.Model):
 __tablename__ = ‘goods_desc‘

 goods_id = db.Column(db.Integer, primary_key=True)
 goods_desc = db.Column(db.Text, default=None)

 def __str__(self):
 return "GoodsDesc => { goods_id:%d, goods_desc:‘%s‘}" % (
self.goods_id, self.goods_desc)

 __repr__ = __str__

# 代码示例
goods_id = 101
table_index = goods_id%2
table_name = ‘goods_desc_%d‘ % table_index
GoodsDesc.__table__.name = table_name
gd = GoodsDesc.query.filter(GoodsDesc.goods_id == goods_id).first()

# 这样写虽然也能工作，但是是非常危险的，因为GoodsDesc.__table__是静态全局变量，
# 而且不是web程序request级别的，是app context的，非常不安全。

经过N次Google，终于找到解决方案，而且代码写起来也很优雅。代码如下：

class GoodsDesc(object):

 _mapper = {}

 @staticmethod
 def model(goods_id):
 table_index = goods_id%100
 class_name = ‘GoodsDesc_%d‘ % table_index

 ModelClass = GoodsDesc._mapper.get(class_name, None)
 if ModelClass is None:
  ModelClass = type(class_name, (db.Model,), {
  ‘__module__‘ : __name__,
  ‘__name__‘ : class_name,
  ‘__tablename__‘ : ‘goods_desc_%d‘ % table_index,

  ‘goods_id‘ : db.Column(db.Integer, primary_key=True),
  ‘goods_desc‘ : db.Column(db.Text, default=None),
  })
  GoodsDesc._mapper[class_name] = ModelClass

 cls = ModelClass()
 cls.goods_id = goods_id
 return cls

# 外部代码调用如例如下：
# -----------------------

# 新增插入
gdm = GoodsDesc.model(goods_id)
gdm.goods_desc = ‘desc‘
db.session.add(gd)

# 查询
gdm = GoodsDesc.model(goods_id)
gd = gdm.query.filter_by(goods_id=goods_id).first()

flask-sqlalchemy分表解决方案

标签：

热心网友时间：2022-04-08 03:11

首先你需要动态地创建一个表结构：
tb_name = 'tb_test'
pk_col = db.Column('tid', db.Integer, primary_key=True, autoincrement=True)
name_col = db.Column('name', db.String(32), nullable=False)
...
tb = db.Table(tb_name, db.metadata, pk_col, name_col, ...)

然后在数据库中创建这个表：
db.metadata.create_all(bind=db.engine, tables=[tb])

如果你想要有一个Model的话，可以将一个类和table映射一下：
# 根据tb_name构造一个类名
class_name = tb_name.title().replace('_', '') + 'Model'
# 创建类
TbTestModel = type(class_name, (object,), {
'__init__': lambda self, name: setattr(self, 'name', name)
})
# 对应的硬编码
# class TbTestModel(object):
# def __init__(self, name):
# self.name = name
db.mapper(TbTestModel, tb)
TbTestModel.query = db.session.query(TbTestModel)

然后就可以愉快地使用TbTestModel进行查询了：
test = TbTestModel('Tom')
db.session.add(test)
db.session.commit()
TbTestModel.query.filter(TbTestModel.name == 'Tom').first()