小编Pla*_*Fan的帖子

我们需要创建SQLAlchemy类来访问多个外部数据源,这些数据源的数量会随着时间的推移而增加.我们为我们的核心ORM模型使用声明性基础,我知道我们可以使用autoload = True手动指定新的ORM类来自动生成映射.

问题是我们需要能够动态生成它们,如下所示:

from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

stored={}
stored['tablename']='my_internal_table_name'
stored['objectname']='MyObject'

并动态地将它变成这样的东西:

class MyObject(Base):
    __tablename__ = 'my_internal_table_name'
    __table_args__ = {'autoload':True}

我们不希望类持续时间超过打开连接,执行查询,然后关闭连接所需的时间.因此,理想情况下,我们可以将上面"存储"变量中的项目放入数据库中,并根据需要提取它们.另一个挑战是对象名称(例如"MyObject")可能在不同的连接上使用,所以我们不能定义它一次并保持它.

关于如何实现这一点的任何建议将不胜感激.

谢谢...

我遇到了一个相当棘手的问题,经过测试后我发现没有可用的答案就足够了.

我已经看到了各种建议,但似乎没有人能够返回MySQL中auto_increment字段的最后插入值.

我看过一些例子,提到使用session.flush()来添加记录然后检索id.然而,似乎总是返回0.

我也看到过提及session.refresh()使用的例子,但会引发以下错误:InvalidRequestError:无法刷新实例''

我想做的事似乎非常简单,但我似乎无法弄明白这个秘密.

我正在使用声明式方法.

所以,我的代码看起来像这样:

class Foo(Base):
    __tablename__ = 'tblfoo'
    __table_args__ = {'mysql_engine':'InnoDB'}

    ModelID = Column(INTEGER(unsigned=True), default=0, primary_key=True, autoincrement=True)
    ModelName = Column(Unicode(255), nullable=True, index=True)
    ModelMemo = Column(Unicode(255), nullable=True)

f = Foo(ModelName='Bar', ModelMemo='Foo')
session.add(f)
session.flush()

此时,对象f已被推送到DB,并且已自动分配了唯一的主键ID.但是,我似乎无法找到一种方法来获取在一些额外操作中使用的值.我想做以下事情:

my_new_id = f.ModelID

我知道我可以简单地执行另一个查询来根据其他参数查找ModelID,但我不愿意,如果可能的话.

我非常感谢任何有关此问题的解决方案的见解.

我在这里先向您的帮助表示感谢.

我正在寻找一个SQLAlchemy唯一的解决方案,用于将从表单提交收到的字典转换为数据库中的一系列行,每个字段用于提交的每个字段.这是为了处理各种应用程序差异很大的首选项和设置.但是,它很可能适用于创建类似功能的数据透视表.我在ETL工具中看到了这种类型的东西,但我一直在寻找一种直接在ORM中完成它的方法.我找不到任何文件,但也许我错过了一些东西.

例:

从表格提交:{"UniqueId":1,"a":23,"b":"你好","c":"世界"}

我希望它被转换(在ORM中),以便它像这样记录在数据库中:

_______________________________________
|UniqueId| ItemName   | ItemValue     |
---------------------------------------
|  1     |    a       |    23         |
---------------------------------------
|  1     |    b       |    Hello      |
---------------------------------------
|  1     |    c       |    World      |
---------------------------------------

在选择时,结果将被转换(在ORM中)回到来自每个单独值的一行数据.

---------------------------------------------------
| UniqueId  |  a     |     b      |       c       |

---------------------------------------------------
|   1       |  23    |   Hello    |   World       |

---------------------------------------------------

我假设在更新时最好的做法是在事务中包装一个删除/创建,这样就会删除当前记录并插入新记录.

最终的ItemNames列表将保存在一个单独的表中.

完全接受更优雅的解决方案,但是如果可能的话,我们希望远离数据库方面.

我在SQLAlchemy中使用了declarative_base方法.

提前致谢...

干杯,

保罗

当需要返回值(即未在WHERE中使用)时,我正在努力理解在SQLAlchemy中执行子查询的正确语法.

我正在使用声明式方法.

有两种模型使用:

class ProjectInvoices(Base):
    InvoiceID = Column(Integer(unsigned=True), default=0, primary_key=True, autoincrement=True)
    MasterProjectID = Column(Integer(unsigned=True), index=True, nullable=False)
    ExpenseAmount = Column(Numeric(10, 2), default=0)
    HoursAmount = Column(Numeric(10, 2), default=0)
    IsVoid = Column(Boolean, default=0, index=True)
    IsSubmit = Column(Boolean, default=0, index=True)

class ProjectMasters(Base):
    MasterProjectID = Column(Integer(unsigned=True), default=0, primary_key=True, autoincrement=True)
    MasterProjectName = Column(Unicode(255))
    MasterProjectMemo = Column(UnicodeText)
    IsActive = Column(Boolean, default=0, index=True)

查询的要点是通过使用子查询对相关发票求和来确定每个项目的当前发票金额.还有其他原因在子查询中完成,而不仅仅是一个连接,所以我真的需要弄清楚子查询问题.

这是我当前SA查询的一个示例:

sqry = session.query(
  func.sum(
    func.ifnull(ProjectInvoices.ExpenseAmount, 0) 
    + func.ifnull(ProjectInvoices.HoursAmount, 0))).label('billed_total')
).filter(and_(ProjectInvoices.IsVoid == 0, ProjectInvoices.IsSubmit == 1)
).subquery()

result = session.query(
  ProjectMasters.MasterProjectID, 
  ProjectMasters.MasterProjectName, 
  sqry.columns.billed_total.label('invoice_total')
).filter(ProjectMasters.IsActive …

我们正在尝试使用paramiko模块根据需要为任意服务器创建SSH隧道,以便查询远程数据库.我们尝试使用paramiko附带的forward.py演示,但最大的限制是,一旦启动套接字服务器,似乎没有一种简单的方法可以关闭SSH隧道和SSH连接.

我们的限制是我们无法从shell激活它,然后手动终止shell以停止列表器.我们需要打开SSH连接,隧道,通过隧道执行一些操作,关闭隧道,并关闭python中的SSH连接.

我已经看到了对server.shutdown()方法的引用,但目前尚不清楚如何正确实现它.

任何帮助将不胜感激...