标签: sqlalchemy

我正在使用Postgresql与SQLAlchemy,但似乎sqlalchemy在使用子查询时添加行时遇到问题.

在我的示例中,我想更新表中特定标记的计数器.

在SqlAlchemy中,测试运行类如下所示:

class TestRun( base ):
    __tablename__   = 'test_runs'

    id              = sqlalchemy.Column( 'id', sqlalchemy.Integer, sqlalchemy.Sequence('user_id_seq'), primary_key=True )
    tag             = sqlalchemy.Column( 'tag', sqlalchemy.String )
    counter         = sqlalchemy.Column( 'counter', sqlalchemy.Integer )

然后,插入代码应如下所示:

tag = 'sampletag'
counterquery = session.query(sqlalchemy.func.coalesce(sqlalchemy.func.max(TestRun.counter),0) + 1).\
                            filter(TestRun.tag == tag).\
                            subquery()

testrun = TestRun()
testrun.tag = tag
testrun.counter = counterquery

session.add( testrun )
session.commit()

这个问题是,在运行此代码时它会产生一个非常有趣的错误,它正在尝试运行以下SQL查询:

'INSERT INTO test_runs (id, tag, counter) 
    VALUES (%(id)s, 
            %(tag)s, 
            SELECT coalesce(max(test_runs.counter), %(param_1)s) + %(coalesce_1)s AS anon_1 
               FROM test_runs 
               WHERE test_runs.tag = %(tag_1)s)' 
 {'coalesce_1': …

如何在SQLAlchemy中添加postgres数组的大小？

像SQL类型Integer [2]:

column = Column(postgresql.ARRAY(Integer), size=2)

我有这个代码来创建一个数据库:

from sqlalchemy import create_engine
from sqlalchemy import Table, Column, Integer, String, Float, MetaData, ForeignKey
from sqlalchemy.sql import select, and_
from PyQt4 import QtGui, QtCore

class DbUtils(object):
    def __init__(self, db_file = None, parent = None):

        self.db = None
        self.db_connection = None
        self.db_file = str(db_file)

    def db_open(self):
        self.db = create_engine('sqlite:///' + self.db_file)
        self.db_connection = self.db.connect()

    def db_close(self):
        self.db_connection.close()

    def db_create_voltdrop(self):
        metadata = MetaData()

        tb_cable_brands = Table('cable_brands', metadata,
            Column('id', Integer, primary_key=True),
            Column('brand', String)
            )
        tb_cable_types = Table('cable_types', metadata,
            Column('id', Integer, primary_key=True),
            Column('brand_id', …

有没有关于如何为windows设置sqlalchemy的教程？我去了www.sqlalchemy.org,他们没有关于Windows设置的明确说明.当我打开压缩包时,我看到distribute_setup,ez_setup和setup.py等文件,但它没有看到安装sqlalchemy.

我有一个相当简单的python Flask应用程序在Apache2下作为WSGI进程运行.应用程序有一个监听器,使用SQLAlchemy从数据库中检索几行数据并将其作为JSON发送回来

对于MySql连接,我确实有一个全局引擎被重用.

使用JMeter生成一些负载,Apache2进程每5秒就将RAM使用量增加0.5%,并且很快就会耗尽RAM.如果停止生成负载的JMeter,则内存不会被清除.

httpd.conf文件

<VirtualHost *:80>
ServerName xxxxxxxxxx.com

<Directory /var/www/xxxxxxxxxx>
    Order allow,deny
Allow from all
</Directory>

WSGIDaemonProcess BiddingPractice user=www-data group=www-data threads=5
WSGIScriptAlias /flask /var/www/xxxxxxxx/xxxxxxxxxxx.wsgi

<Directory /var/www/BiddingPractice>
    WSGIProcessGroup BiddingPractice
    WSGIApplicationGroup %{GLOBAL}
    Order deny,allow
    Allow from all
</Directory>

WSGI文件

import sys
sys.path.insert(0, '/home/stefan/Code/xxxxxx')
from BiddingPractice import app as application

_ init.py _

    # -*- coding: utf-8 *-*
from flask import Flask
from sqlalchemy import *

app = Flask(__name__)
db = create_engine('mysql://root:xxxxxx@localhost/xxxxxxx')

import BiddingPractice.main

main.py

    # -*- coding: utf-8 *-*
from flask …

我需要一些帮助来理解继承在SQLAlchemy中的工作原理.我为具有一些基本功能的用户创建了一个基类.然后是一些特定用户(admin,cooluser,uncooluser).每个都有独特的功能,所以我决定在SQLAlchemy中使用继承.问题是我需要能够将用户升级到cooluser或uncooluser,并在任何时候将cooluser降级为用户.

class User(Base):
    __tablename__ = 'tbl_users'
    __table_args__ = {'mysql_engine': 'InnoDB'}

    user_id = Column(Integer, primary_key = True, unique = True, nullable = False)
    user_name = Column(String(100), nullable = False, unique = True)
    password = Column(String(100), nullable = False)
    user_type = Column('user_type', String(50))
    first_name = Column(String(50), nullable = False)
    last_name = Column(String(50), nullable = False)
    address = Column(String(50), nullable = False)
    city = Column(String(20), nullable = False)
    postal_code = Column(String(10), nullable = False)
    country_id = Column(Integer, ForeignKey(Contries.country_id))

    country = relationship('Country', backref = …

我试图提高一些查询性能,但生成的查询看起来并不像我期望的那样.

使用以下方法检索结果:

query = session.query(SomeModel).
        options(joinedload_all('foo.bar')).
        options(joinedload_all('foo.baz')).
        options(joinedload('quux.other'))

我想要做的是过滤通过'first'连接的表格,但这种方式不起作用:

query = query.filter(FooModel.address == '1.2.3.4')

它会在查询中附加一个这样的子句:

WHERE foos.address = '1.2.3.4'

其中没有以适当的方式进行过滤,因为生成的连接附加表foos_1和foos_2.如果我手动尝试该查询,但将过滤子句更改为:

WHERE foos_1.address = '1.2.3.4' AND foos_2.address = '1.2.3.4'

它工作正常.问题当然是 - 我如何通过sqlalchemy本身实现这一目标？

我有三个表定义为:

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer(10), primary_key=True)
    firstname = Column(String(64))
    surname = Column(String(64))

class SWMS(Base):
    __tablename__ = 'swms'
    id = Column(Integer(10), primary_key=True)
    owner_id = Column(Integer(10), ForeignKey('users.id', ondelete='CASCADE'))
    filename = Column(String(255))

    swmowner = relationship('User', backref=backref('users'))

class SWM_perms(Base):
    __tablename__ = 'swm_perms'
    id = Column(Integer(10), primary_key=True)
    swm_id = Column(Integer(10), ForeignKey('swms.id', ondelete='CASCADE'))
    user_id = Column(Integer(10), ForeignKey('users.id', ondelete='CASCADE'))

    swm = relationship('SWMS', backref=backref('swms'))
    swmuser = relationship('User', backref=backref('swmusers'))

本质上,SWMS表是文档信息表,其中owner_id定义创建文档的用户.SWM_perms是一个具有文档ID到用户ID映射的表 - 用于定义允许哪些用户查看文档.

要生成一个表,其中包含1)用户拥有的所有文档,或2)用户可以查看,我会这样做:

select owner_id, users.firstname, users.surname, filename
from swms, swm_perms, users
where users.id=swms.owner_id and …

我有一个Sybase表,使用sqlalchemy进行声明映射,如下所示:

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

class Appointment(base):
    __tablename__ = 'APPOINTMENT'
    __table_args__ = {'quote':False}

    dt = Column(Date, name='dt_appointment')

如何查询此表,按年份排序(dt_appointment)？Year()是一个有效的sybase T-SQL函数.我试过包括一个计算列,例如:

dtYear = Column(Integer, name='Year(dt_appointment)', quote=False)

这不适用于查询:

session.Query(Appointment).order_by(Appointment.dtYear)

尝试使用Python的csv.writer生成的bcp文件时,我遇到了EOF问题.我做了大量的谷歌搜索,没有运气,所以我转向你有用的人在SO

这是错误消息(在subprocess.call()行上触发):

Starting copy...
Unexpected EOF encountered in BCP data-file.
bcp copy in failed

这是代码:

sel_str = 'select blahblahblah...'
result = engine.execute(sel_str)  #engine is a SQLAlchemy engine instance

# write to disk temporarily to be able to bcp the results to the db temp table
with open('tempscratch.csv','wb') as temp_bcp_file:
    csvw = csv.writer(temp_bcp_file)
    for r in result:
        csvw.writerow(r)
        temp_bcp_file.flush()

# upload the temp scratch file
bcp_string = 'bcp tempdb..collection in @INFILE -c -U username -P password -S DSN'
bcp_string = string.replace(bcp_string,'@INFILE','tempscratch.csv') …