标签: sqlalchemy

这是我当前模型的简化示例(我正在使用Flask SQLAlchemy扩展):

like = db.Table(
    'like',
    db.Column('uid', db.Integer, db.ForeignKey('users.id')),
    db.Column('pid', db.Integer, db.ForeignKey('posts.id'))
)

class User(db.Model):
    __tablename__ = 'users'

    id = db.Column(db.Integer, primary_key = True)
    username = db.Column(db.String(20))

class Post(db.Model):
    __tablename__ = 'posts'

    id = db.Column(db.Integer, primary_key = True)
    title = db.Column(db.String(255))

    likes = db.relationship(
        'User',
        secondary = like,
        backref = db.backref('likes', lazy = 'dynamic'),
        lazy = 'dynamic'
    )

我试图Post通过喜欢的数量订购.

这是我基本上试图发出的查询:

SELECT p.*, COUNT(l.`pid`) as `likes`
FROM `posts` as p
LEFT JOIN `like` …

我需要使用sqlalchemy在mysql中创建一个db,如果它已经存在,我可以连接到db,但是我希望能够创建它,如果它不存在的话.这是我的表:

    #def __init__(self):
Base = declarative_base()

class utente(Base):
    __tablename__="utente"
    utente_id=Column(Integer,primary_key=True)
    nome_utente=Column(Unicode(20))
    ruolo=Column(String(10))
    MetaData.create_all()

    def __repr(self):
        return "utente: {0}, {1}, id: {2}".format(self.ruolo,self.nome_utente,self.utente_id)


class dbmmas(Base):

    __tablename__="dbmmas"
    db_id=Column(Integer,primary_key=True,autoincrement=True)
    nome_db=Column(String(10))
    censimento=Column(Integer)
    versione=Column(Integer)
    ins_data=Column(DateTime)
    mod_data=Column(DateTime)
    ins_utente=Column(Integer)
    mod_utente=Column(Integer)
    MetaData.create_all()

    def __repr(self):
        return "dbmmas: {0}, censimento {1}, versione {2}".format(self.nome_db,self.censimento,self.versione)    

class funzione(Base):
    __tablename__="funzione"
    funzione_id=Column(Integer,primary_key=True,autoincrement=True)
    categoria=Column(String(10))
    nome=Column(String(20))
    def __repr__(self):
        return "funzione:{0},categoria:{1},id:{2} ".format(self.nome,self.categoria,self.funzione_id)

class profilo(Base):
    __tablename__="rel_utente_funzione" 
    utente_id=Column(Integer,primary_key=True)
    funzione_id=Column(Integer,primary_key=True)
    amministratore=Column(Integer)
    MetaData.create_all()

    def __repr(self):
        l=lambda x: "amministratore" if x==1 else "generico"
        return "profilo per utente_id:{0}, tipo: {1}, funzione_id: {2}".format(self.utente_id,l(self.amministratore),self.funzione_id)    

class aree(Base):
    __tablename__="rel_utente_zona" …

我正在使用SQlAlchemy.我想删除有效存在于数据库中的所有记录,但我不想删除表/数据库.

我尝试使用以下代码.

con = engine.connect()
trans = con.begin()

con.execute(table.delete())
trans.commit()

看起来,它不是非常有效的,因为我迭代数据库中存在的所有表.有人可以建议我这样做更好,更有效吗？

任何形式的帮助在这里都非常感激.

对于SQLAlchemy的,谁可以轻轻地得到的简单的例子SQL类似功能sum,average,min,max,对于一个柱(score在作为示例的以下).

至于这个映射器:

class Score(Base):
    #...
    name = Column(String)
    score= Column(Integer)
    #...

我正在学习sqlalchemy.这是我的初始代码:

文件:user.py

from sqlalchemy import Column,Integer,Sequence, String
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class User(Base):
  __tablename__ = 'users'
  id = Column(Integer,Sequence('user_seq'),primary_key=True)
  username   = Column(String(50),unique=True)
  fullname = Column(String(150))
  password  = Column(String(50))
  def __init__(self,name,fullname,password):
    self.name = name
    self.fullname = fullname
    self.password = password

文件main.py

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from user import User
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
if __name__ == '__main__':
  engine = create_engine('mysql://root:password@127.0.0.1:3306/test',echo=True)
  Base.metadata.create_all(engine, checkfirst=True)
  Session = sessionmaker(bind=engine)
  session = Session()
  ed_user = User('ed', …

我有各种类似(但不完全相同)的数据库,并希望使用SQLAlchemy作为"标准化"访问的方式.数据库可以略有不同,例如在列名称上具有唯一的前缀,或者它们可以更显着地区别并且缺少列(或者对于旧数据库,缺少整个表).

我正在寻求帮助的不是SQLAlchemy问题,而是Python /组织问题.如何才能轻松设置多个可在项目中重复使用的数据库？

我已经阅读了关于SQLAlchemy会话的内容,但是在没有实例化每个项目的情况下都看不到使用它们的方法.

我的问题是:我如何制作一个模块/包,其中包含许多数据库模型设置,可以在SQLAlchemy中使用,可以在另一个python项目中轻松导入/使用？

我还没有那么担心处理丢失的列/表.我可以稍后解决这个问题,但需要记住这一点,因为我不能为每个数据库使用完全相同的模型.

关于这个主题的任何资源,指针或阅读材料都将得到真正的赞赏.在此先感谢,如果在其他地方已经回答,我很抱歉,搜索没有显示与此相关的任何内容.

编辑:我原来保留原文,并根据保罗的建议添加更多内容.

RE:SA ORM - 是的,我计划使用SQLAlchemy ORM.由于可能是显而易见的原因,我无法提供真正的数据库.但是,假设这三个虚构的数据库,恰当地命名为DB1,DB2和DB3(我们假设每个数据库中只有一个表,只有几列,现实世界中两者都要多得多).

每个数据库都有一个用户表,每个用户表中都有几列.以下是表/列的一些SQL表示法:

DB1.user  --> DB1.user.id,      DB1.user.username,  DB1.user.email
DB2.user  --> DB2.user.id,      DB2.user.user_name, DB2.user.email
DB3._user --> DB3._user.userid, DB3._user.username, DB3.user.email_address

目前,我正在尝试将这些数据库分离为"模块化",并且能够随时添加其他数据库.

我考虑了几个不同的文件组织方面(假设__init__.py存在于需要的地方,但为了简洁而省略),包括:

Databases         |    Databases            |    Databases
    DB1.py        |        DB1              |        DB1
    DB2.py        |            models.py    |            models
    DB3.py        |        DB2              |                user.py
                  |            models.py    |                anothertable.py
                  |        DB2              |        ...
                  |            models.py    |        DB3
                  |                         |            models
                  |                         |                user.py
                  |                         |                anothertable.py

我希望能够使用SA ORM访问这些,并且当在python文件中使用这些数据库时,尽可能少地导入/声明.需要做类似的事情:

from sqlalchemy import create_engine
from …

这是我的简单测试脚本.只是尝试做一个基本的选择语句.找到教程的基本部分.

from sqlalchemy import *

db = create_engine('mssql+pyodbc://user:pass@ip_address/database_name')    

db.echo = True 
metadata = MetaData(db)

users = Table('member', metadata, autoload=True)

def run(stmt):
    rs = stmt.execute()
    for row in rs:
        print row

s = users.select(users.c.fname == 'Bill')
run(s)

经过一个小时的搜索并尝试了一些解决方案之后,我就没有比开始时更接近解决它了.希望我在某个地方犯了一个简单的错误,但我找不到它......

这是我得到的错误

sqlalchemy.exc.DBAPIError: (Error) ('IM002', '[IM002] [unixODBC][Driver Manager]Data source name not found, and no default driver specified (0) (SQLDriverConnect)') None None

任何帮助将非常感激!

我正在尝试运行alembic迁移,当我运行时

alembic revision --autogenerate -m "Added initial tables"

它说不出来

sqlalchemy.exc.ArgumentError: Can't load plugin: sqlalchemy.dialects:driver

数据库网址是

postgresql+psycopg2://dev:passwd@localhost/db

我甚至psycopg2安装在我的virtualenv中

$yolk -l
Flask-Login     - 0.1.3        - active
Flask-SQLAlchemy - 0.16         - active
Flask           - 0.9          - active
Jinja2          - 2.6          - active
Mako            - 0.7.3        - active
MarkupSafe      - 0.15         - active
Python          - 2.7.2        - active development (/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/lib-dynload)
SQLAlchemy      - 0.8.0        - active
Werkzeug        - 0.8.3        - active
alembic         - 0.4.2        - active
antiorm         - 1.1.1        - active …

我可以alembic --autogenerate在添加/删除列时使用 .

但是,当我想修改例如200个字符到2000个字符的"url"列时,它不会检测到更改.

如何创建Alembic(使用SQLAlchemy),检测更改并自动生成脚本到我的模型的各种列的"大小",并为PostgreSQL创建"alter_column"命令？

编辑:

为什么不使用alembic自动添加:

op.alter_column('mytable', 'url', type_=sa.String(2000), existing_type=sa.String(length=200), nullable=True)

我有一个Flask应用程序,它使用Flask-SQLAlchemy连接到MySQL数据库.

我希望能够检查表中是否存在行.我如何修改这样的查询以检查行是否存在:

db.session.query(User).filter_by(name='John Smith')

我找到了一个使用SQLAlchemy的问题的解决方案,但似乎不符合Flask-SQLAlchemy的工作方式:

from sqlalchemy.sql import exists    
print session.query(exists().where(User.email == '...')).scalar()

谢谢.