标签: sqlalchemy

我正在寻找动态查询对象的属性。我不知道在这种情况下我将在执行时使用哪个属性或列。

class Product(Base):
    __tablename__ = 'products'

    sku = Column(String, primary_key=True)
    list_price = Column(String)
    status = Column(String)
    url = Column(String)
    special_price1 = Column(String)
    special_price2 = Column(String)
    special_price3 = Column(String)

我有一个 SQLAlchemy 基类Product，它描述了一些属性，以及与标价不同的其他特殊价格。

然后我PriceList在下面有一个类，它可以访问其他资源和方法，这些资源和方法有助于报告和更新表中的'products'列。此类存储有关所有Product对象的唯一特价清单的信息。

class PriceList:

    def __init__(self, name, db_col_name):
        # Display name
        self.name = name

        # Used for querying the database for a specific column
        # This will always be one of the 4 price related column names
        # list_price, special_price1, special_price2, or …

我正在使用 Pandas 从我的 sql server 读取表，例如

df= pd.read_sql('table1', engine)

引擎是我的 pyodbc 连接，然后我再次将它推送到 sql server

df.to_sql('table2', engine, if_exists='replace')

这给了我一个错误

ValueError: duplicate name in index/columns: cannot insert level_0, already exists

当我尝试删除该列时，它又给了我一些错误，这无论如何都不是一种有效的方法。我也试过这个，这也不起作用

 df= df.reset_index(drop=True)

每一个帮助都很重要

如何查看实际插入了多少行？

engine_str = 'mysql+mysqlconnector://root:mypass@localhost/mydb'
engine = sqlalchemy.create_engine(engine_str, echo=False, encoding='utf-8')
connection = engine.connect()
query = "INSERT INTO TEST(a,b,c)VALUES(1,2,3) ON DUPLICATE KEY UPDATE B = VALUES(B), C = VALUES(C)"
try:
    connection.execute(query)
except Exception as e:
    print(e)

SQLALCHEMY 只是打印出这个 ID。

<sqlalchemy.engine.result.ResultProxy object at 0x0000000007769E80>

PS 这个问题是关于行插入而不是行执行。因此，如果您使用 rowcount，它将显示查询已执行的次数。

重新表述我的问题：有没有办法查看插入和更新的记录数？理想情况下，我想知道那些更新的主键。

例外是AttributeError: 'StringField' object has no attribute 'wrap_formdata'，它似乎只发生在我从 Flask-Admin 仪表板创建或编辑以下两个模型中的任何一个时：

class Experiment(db.Model):
    id = db.Column(db.Integer, primary_key=True)

    property_id = db.Column(db.Integer,
                            db.ForeignKey('property.id'),
                            index=True,
                            nullable=False)

    name = db.Column(db.String, nullable=False)

    start_date = db.Column(db.DateTime, nullable=True)
    end_date = db.Column(db.DateTime, nullable=True)
    status = db.Column(db.String, nullable=False, default='Draft')

    title = db.Column(db.String)
    meta = db.Column(db.String)
    seed_uri = db.Column(db.String)

    targets = db.Column(postgresql.JSON)
    variations_json = db.Column(postgresql.JSON)

    variations = db.relationship('ExperimentVariation',
                                  backref='experiment',
                                  cascade='save-update, merge, delete')

    def __repr__(self):
        repr_fmt = '<Experiment {id}, {property_id} {name}>'
        return repr_fmt.format(id=self.id,
                               property_id=self.property_id,
                               name=self.name)


class ExperimentVariation(db.Model):
    id = db.Column(db.Integer, primary_key=True) …

我们有一个db.py定义了 peewee 数据库的文件：

db = PostgresqlExtDatabase('mom',
                           user=DB_CONFIG['username'],
                           password=DB_CONFIG['password'],
                           host=DB_CONFIG['host'],
                           port=DB_CONFIG['port'],
                           threadlocals=True,
                           register_hstore=False,
                           autocommit=True,
                           autorollback=True,
                           cursor_factory=DictCursor)

调用db.execute("SOME RAW SQL UPDATE QUERY")按预期工作。但是begin在此之前调用 a并不会阻止数据库被修改。

db.begin()
db.execute("SOME RAW SQL UPDATE QUERY")  # <- Does not wait, db is updated immediately here
db.commit()

我这样做对吗？

如果一个事务已经在进行中，我基本上需要将原始 sql 嵌套在一个事务中，否则如果没有begin调用任何事务，则立即执行它。

这工作，如果我做的预期db.set_autocommit(False)，然后execute_sql再commit()。
它也适用于atomic()上下文管理器。

给予一定的情况下，我工作的一个Web应用程序，物流，和我们的代码库使用瓶和SQLAlchemy的scoped_session使用autocommit设置为true。它不使用SQLAlchemy的ORM（因为..历史原因），而是只使用Session对象和它的 execute()，begin()，begin_nested()，rollback()和remove()方法。

它的实现方式是Session = …

所以我正在尝试使用数据库和烧瓶动态构建站点。该网站现在以我希望的方式工作，但只持续了大约一分钟，然后我收到一条错误消息：

sqlalchemy.exc.OperationalError: (pymysql.err.OperationalError) (2013, 'Lost connection to MySQL server during query')

数据库包含我想为每一页抓取的 4 个条目，以及一个主页，我只从数据库中的每个条目抓取 2 个条目。沿着这条线的某个地方，它遇到了一个错误，它无法再访问数据库，因此不允许加载更多使用数据库的页面。

谁能帮我弄清楚为什么会发生这种情况？我对烧瓶和 sqlalchemy 很陌生

下面是我的代码：

数据库.py

from flask import Flask
from flask import render_template
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://databaseaddress'
db = SQLAlchemy(app)


class Kitchen(db.Model):
    __tablename__ = 'kitchens'
    id = db.Column('id', db.Integer, primary_key=True)
    description = db.Column('description')
    galleryId = db.Column('galleryId')
    cpo = db.Column('cpo')

    def __init__(self, description, galleryId, cpo):
        self.description = description
        self.galleryId = galleryId
        self.cpo = cpo

    def __repr__(self):
        return '<Gallery %r>' …

我正在研究 ResultProxy 类，它似乎是该execute方法的结果。

我从一个简单的选择开始，然后对结果进行迭代。我很好奇的是更新和删除。我怎么知道删除是否成功？

在 python 中这样做时，我会有类似的东西：

result = session.execute("delete from A where id = 4")

它运行它，但除了尝试查询 id=4 的存在之外，我不知道它是否成功。

当我查看 ResultProxy 类时，我没有看到任何与我相关的内容isSuccessful或任何其他跳出给我的关键字。

就像我说的，通常，我只是遍历选择集，但无法弄清楚这一点。

我有一个包含三个表（A、B 和 C）的 sqlalchemy 模式，通过一对多外键关系（在 A->B 之间）和（B->C）与 SQLite 作为后端相关。我创建了单独的数据库文件来存储数据，每个文件都使用完全相同的 sqlalchemy 模型并运行相同的代码将数据放入其中。

我希望能够从所有这些单独的数据库中复制数据并将它们放入一个新的数据库文件中，同时保留外键关系。我尝试了以下代码将数据从一个文件复制到一个新文件：

import sqlalchemy
from sqlalchemy.ext import declarative
from sqlalchemy import Column, String, Integer
from sqlalchemy import orm, engine

Base = declarative.declarative_base()
Session = orm.session_maker()

class A(Base):
    __tablename__ = 'A'

    a_id = Column(Ingeter, primary_key=True)
    adata = Column(String)
    b = orm.relationship('B', back_populates='a', cascade='all, delete-orphan', passive_deletes=True)


class B(Base):
    __tablename__ = 'B'

    b_id = Column(Ingeter, primary_key=True)
    a_id = Column(Integer, sqlalchemy.ForeignKey('A.a_id', ondelete='SET NULL') 
    bdata = Column(String)
    a = orm.relationship('A', back_populates='b')
    c = orm.relationship('C', back_populates='b', …

我现在正在尝试在 SQLAlchemy 中从旧课程中实现非常简单的示例表...

我已经走了这么远但是当我运行代码时......

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Date, MetaData
from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy import create_engine

Base = declarative_base()
engine = create_engine('mysql://x @ amazonaws.com:3306/db', echo=True)
class Guest(Base):
    __tablename__ = "guests"
    guest_no = Column(String(4), primary_key=True)
    g_name = Column(String(20))
    g_address = Column(String(30))
    booking = relationship("Booking", back_populates="guests")
class Hotel(Base):
    __tablename__ = "hotels"
    hotel_no = Column(String(4), primary_key=True)
    h_name = Column(String(20))
    h_address = Column(String(30))
    room = relationship("Room", back_populates="hotels")
    booking = relationship("Booking", back_populates="hotels")
class …

我在数据库中有很多重复项。为了只提供一个（而不是重复项），我正在尝试使用set，但它也返回了重复项。

all_orders = Order.query.filter_by(account_id=account.id).all()
orders = set(all_orders)

{377, 348, 353, 377, 354, 377, 356, 377, 357, 377, 359, 378, 358, 378, 361, 378, 357, 378, 364, 378, 363, 378, 378, 362, 364, 379, 355, 379, 366, 379, 367, 379, 368, 379, 358, 379, 369, 379, 358, 379, 369, 379, 337, 370, 379, 338, 370, 379, 339, 343, 371, 379, 347, 371, 379, 372, 379, 372, 379, 373, 379, 373, 379, 374, 379, 374, 379, 375, 379, 375, 379, …