标签: sqlalchemy

有两个(三个,但我不算Elixir,因为它不是"官方")用SQLAlchemy定义持久化对象的方法:

映射器对象的显式语法

from sqlalchemy import Table, Column, Integer, String, MetaData, ForeignKey
from sqlalchemy.orm import mapper

metadata = MetaData()

users_table = Table('users', metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String),
)

class User(object):
    def __init__(self, name):
        self.name = name

    def __repr__(self):
       return "<User('%s')>" % (self.name)

mapper(User, users_table) # &lt;Mapper at 0x...; User&gt;

声明性语法

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

class User(Base):
     __tablename__ = 'users'
     id = Column(Integer, primary_key=True)
     name = Column(String)

     def __init__(self, name):
         self.name = name

     def …

我是一个SQLAlchemy noob.

假设我在声明模式下有一个用户表:

class User(Base):
    __tablename__ = 'user'
    id = Column(u'id', Integer(), primary_key=True)
    name = Column(u'name', String(50))

当我知道用户的id没有加载到会话中的对象时,我更新这样的用户:

ex = update(User.__table__).where(User.id==123).values(name=u"Bob Marley")
Session.execute(ex)

我不喜欢使用User.__table__,我应该不再担心吗？

有一个更好的方法吗？

谢谢!

我有一个使用sqlalchemy(在Pylons内)的Web应用程序.我需要有效地更改架构,以便至少每天更改生产版本,可能更多,而不会丢失数据.

我在周末用sqlalchemy-migrate进行了一些玩法,我会说这给我留下了不好的印象.首先,我认为它无法帮助两个数据库引擎之间的迁移 ; 这可能只能通过sqlalchemy完成.其次,文档似乎不是最新的.我不得不更改一些命令行选项,比如在每个命令中给出存储库路径,这可能是迁移的错误.

但最糟糕的是它是"manage.py test "命令.它不仅实际修改了数据库(文档中明确指出了这一点,所以我不能责怪迁移),但是我的第一个迁移脚本只是进行了简单的模式迁移,使升级后的数据库与原始数据库的模式不同.但是"manage.py测试"只回答了类似的问题

 success !

也就是说,它甚至没有检查架构是否处于连贯状态.那么迁移是否值得？与S.Lott提出的良好实践相关的"自己动手"方法相比,是否有任何优势？是否有sqlalchemy-migrate的替代方案实际上简化了迁移过程,或者我只是尝试使用先验不好的迁移(那么请告诉我为什么不明显优于创建上面链接中提出的CSV列)？

非常感谢!

我在Python中使用SQLAlchemy,我想知道如何获取列中的总行数.我定义了变量:

engine = sqlalchemy.create_engine(url, ehco=False)
Session = sqlalchemy.orm.sessionmaker(bind=engine)
Session = session()
metadata = sqlalchemy.MetaData(engine)
Base = declarative_base(metadata=metadata)

# A class representing the shape_congress_districts_2012 table
class Congress(Base):
    __tablename__ = 'shape_congress_districts_2012'
    id = geoalchemy.Column(sqlalchemy.Integer, primary_key=True)
    name = geoalchemy.Column(sqlalchemy.Unicode)
    geom = geoalchemy.GeometryColumn(geoalchemy.Polygon(2))
    geom_simple = geoalchemy.GeometryColumn(geoalchemy.Polygon(2))
    area = geoalchemy.Column(sqlalchemy.Float)
    state_id = geoalchemy.Column(sqlalchemy.Integer)
    census_year = geoalchemy.Column(sqlalchemy.Date)

geoalchemy.GeometryDDL(Congress.__table__)

我想确定表中的总行数,而不必等待一大堆时间查询数据库.目前,我有一些代码:

rows = session.query(Congress).all()

然后我可以从列表中访问它们,但这需要我立即将所有内容加载到内存中.

尝试使用to_sql将pandas数据帧写入MySQL表.以前一直在使用flavor ='mysql',但是它将来会被折旧并且想要开始转换到使用SQLAlchemy引擎.

示例代码:

import pandas as pd
import mysql.connector
from sqlalchemy import create_engine

engine = create_engine('mysql+mysqlconnector://[user]:[pass]@[host]:[port]/[schema]', echo=False)
cnx = engine.raw_connection()
data = pd.read_sql('SELECT * FROM sample_table', cnx)
data.to_sql(name='sample_table2', con=cnx, if_exists = 'append', index=False)

读取工作正常,但to_sql有错误:

DatabaseError:sql上的执行失败'SELECT name FROM sqlite_master WHERE type ='table'AND name =？;':字符串格式化过程中参数数量错误

为什么看起来它试图使用sqlite？sqlalchemy与mysql,特别是mysql.connector的正确使用是什么？

我也尝试将引擎作为连接传递,这给了我一个引用没有游标对象的错误.

data.to_sql(name='sample_table2', con=engine, if_exists = 'append', index=False)
>>AttributeError: 'Engine' object has no attribute 'cursor'

我只想以编程方式确定SQLalchemy模型的主键名称.

在Python中,使用SQLAlchemy,我想插入或更新一行.我试过这个:

existing = db.session.query(Toner)
for row in data:
    new = Toner(row[0], row[1], row[2])

这是行不通的.如何new在Toner表中插入或更新？我怀疑它已经完成了合并,但我无法理解如何做到这一点.

我正在使用Flask在论坛模板上工作.当我尝试使用表单在浏览器中创建新线程时,SQLAlchemy会抛出AttributeError.当我尝试与Forum-to-Thread实现一对多的关系以及与Thread-to-User的一对多关系时,问题出现了.

models.py

class User(db.Model):
  id = db.Column(db.Integer, primary_key=True)
  username = db.Column(db.String(32), index=True, unique=True)
  password = db.Column(db.String(32), index=True)
  email = db.Column(db.String(120), index=True, unique=True)
  role = db.Column(db.SmallInteger, default=ROLE_USER)

  posts = db.relationship('Post', backref='author', lazy='dynamic')

class Forum(db.Model):
  id = db.Column(db.Integer, primary_key=True)
  title = db.Column(db.String(128))
  description = db.Column(db.Text)

  threads = db.relationship('Thread', backref='forum', lazy='dynamic')

class Thread(db.Model):

  id = db.Column(db.Integer, primary_key=True)
  title = db.Column(db.String(128))
  author= db.Column(db.String(32))
  timestamp = db.Column(db.DateTime)
  forum_id = db.Column(db.Integer, db.ForeignKey('forum.id'))

  posts = db.relationship('Post', backref='thread', lazy='dynamic')

class Post(db.Model):

  id = db.Column(db.Integer, primary_key=True)
  body = db.Column(db.Text) …

如何使SQLAlchemy中Tornado是async？我在async mongo示例中找到了MongoDB的示例,但我找不到类似motor的内容SQLAlchemy.有没有人知道如何SQLAlchemy执行查询tornado.gen(我在MySQL下面使用SQLAlchemy,目前我的处理程序从数据库读取并返回结果,我想使这个异步).

我无法使用SQLAlchemy创建单个表.

我可以通过调用来创建它,Base.metadata.create_all(engine)但随着表的数量增加,这个调用需要很长时间.

我动态创建表类,然后填充它们.

from sqlalchemy import create_engine, Column, Integer, Sequence, String, Date, Float, BIGINT
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()

class HistoricDay():

    id = Column(Integer, Sequence('id_seq'), primary_key=True)
    #  Date, Open, High, Low, Close, Volume, Adj Close
    date = Column(Date)
    open = Column(Float)
    high = Column(Float)
    low = Column(Float)
    close = Column(Float)
    volume = Column(BIGINT)
    adjClose = Column(Float)

    def __init__(self, date, open, high, low, close, volume, adjClose):
        self.date = date
        self.open = open
        self.high …