标签: sqlalchemy

请注意，这是针对核心的，而不是针对 orm 的。

希望有人能帮助我解决这两个问题：

1）看起来很outerjoin简单join，但是如何进行内连接呢？

2）进行多重连接的语法是什么。我能够进行一次连接，但不确定多个连接的语法。

我的第一个连接看起来像这样：

select([...]).select_from(outerjoin(a, b))

但此语法执行两个连接时会产生一些错误：

select([...]).select_from(outerjoin(a, b).select_from(outerjoin(ma, tr))

提前致谢。

我正在使用 sqlalchemy 的分页对我的查询进行分页，如下所示，但在我的首页中，我想实现下一个按钮（或数字页面导航链接，如 1,2,3,4..），当它到达最后一页时我不想向用户显示下一个按钮（或知道页面导航中可用的最大页面数）。我不想在 sqlalchemy 中使用另一个数据库查询。实现这一目标最方便的方法是什么？

我正在使用以下分页：

Blog.query.filter(Blog.title.like("%"+query+"%")).paginate(page=start,per_page=size).items

一种方法可能是检查它是否返回与我大小相同的项目的确切数量，或者不是我现在正在做的事情，如果不是，它就是最后一个，但它不满足所有计数和大小的余数为零的极端情况。

大家，

关于使用flask-sqlalchemy按年/月搜索..

有没有Pythonic和简短的方法来按年/月搜索？

myquery = Table.query.filter(Table.dt.year == 2016, Table.dt.month == 10)

（这其实是不正确的，只是为了表达我的逻辑）

其中 Table 和 dt 是这样的：

class Table(db.Model):

    ...
    dt = db.Column(db.Datetime)

我正在尝试根据一些过滤器更新表中的行。我可以获得，如果行数为一，并且在保存会话时row_object抛出的行数超过一，我可以更新它。StaleDataError我的数据库中有 2 行

以下是代码..

@require_context
def travel_update(context, travel_id, region, resource):
    with write_session() as session:
        time_now = timeutils.utcnow()

         travel_ref = session.query(models.Region_travel). \
             filter_by(id=travel_id, region=region, resource=resource). \
             first()
        if not travel_ref:
            raise exception.idNotFound(travel_id=travel_id)
        travel_ref.status = "failed"
        travel_ref.updated_at = time_now
        travel_ref.save(session)
        return travel_ref

看起来 SQLAlchemy 默认将VARCHAR2列创建为CHAR. 我怎样才能让它创建呢BYTE？

from sqlalchemy import MetaData, Column, String
from sqlalchemy.ext.declarative import declarative_base

metadata = MetaData()
Base = declarative_base(metadata=metadata)

class Foo(Base):
    __tablename__ = 'foo'
    name = Column(String(10), primary_key=True)

Foo.__table__.create(bind=engine)

这将创建下表：

from sqlalchemy import MetaData, Column, String
from sqlalchemy.ext.declarative import declarative_base

metadata = MetaData()
Base = declarative_base(metadata=metadata)

class Foo(Base):
    __tablename__ = 'foo'
    name = Column(String(10), primary_key=True)

Foo.__table__.create(bind=engine)

相反，我希望它创建以下内容：

CREATE TABLE XXMD.FOO
(
  NAME  VARCHAR2(10 CHAR)                       NOT NULL
)

我希望在可能不超过 10 分钟左右的时间内将大小不超过 50MB 的 .csv 文件批量插入 Azure SQL 数据库中。

但是，批量插入不适用于云。正如另一个线程中指出的，BCP 是一个可行的解决方案，但它需要将数据批量插入本地数据库。然而，这在 PaaS 基础设施中不是一个选项，因为在操作系统级别没有控制，并且无法创建和管理数据库。

目前，使用标准 S3：100 DTU、250 GB Azure SQL 数据库逐个上传 131KB 的 .csv 需要 10 分钟以上。如何将上传速度提升至少一个数量级？

import pandas as pd
import sqlalchemy
table = pd.Dataframe(<your_data_here>)
sql_driver = <driver>
sql_server = <server>
sql_user = <user>
sql_pass = <password>
sql_db = <database_name>
sql_conn = sqlalchemy.create_engine(
    'mssql+pyodbc://'+sql_user+':'+sql_pass+'@'+sql_server+'/'+sql_db+'?driver='+sql_driver)
table.to_sql('table_name', con=sql_conn)

所以当我 select * from table1 t1 left outer join table2 t2 on t1.id = t2.id;在sqlite3终端运行时

我按照我的意愿和预期取回了数据。

但是，当我在 SqlAlchemy 中运行它时 TableOneModel.query.outerjoin(TableTwoModel,TableOneModel.id == TableTwoModel.id)

我只得到table1信息。我什至没有从 中得到空列table2。我错过了一些愚蠢的事情吗？

我的 Flask-admin 应用程序中有三个相关的 SQLAlchemy 模型（简化）：

class Client(db.Model, BasicMixin, ActiveMixin, TimestampMixin):
  id = db.Column(UUIDType, primary_key=True, default=uuid.uuid4)
  title = db.Column(db.String(1000))
  issues = db.relationship('Issue', backref='client', cascade='all, delete-orphan')


class Issue(db.Model, BasicMixin, ActiveMixin, TimestampMixin):
  id = db.Column(UUIDType, primary_key=True, default=uuid.uuid4)
  date = db.Column(db.Date, default=datetime.date.today())
  client_id = db.Column(UUIDType, db.ForeignKey('clients.id'), nullable=False)
  articles = db.relationship('Article', backref='issue', cascade='all, delete-orphan')


class Article(db.Model, BasicMixin, TimestampMixin):
  id = db.Column(UUIDType, primary_key=True, default=uuid.uuid4)
  title = db.Column(db.String())
  body = db.Column(db.String())
  issue_id = db.Column(UUIDType, db.ForeignKey('issues.id'), nullable=False)

客户有很多Issue，每个Issue有很多文章。

我还有一个文章的模型视图，其中我应该能够按客户过滤文章列表（按名称选择客户并仅显示属于该客户的文章）。我应该怎么做才能在flask-admin 中创建这样的过滤器？

我正在创建一个应用程序来替换浏览器中当前的选项卡管理器。我在组 - 表主题和选项卡 - 表选项卡之间创建了一个简单的一对多关系。我希望在删除主题时能够自动删除它的子主题。这就是我目前所拥有的：

from flask import request, redirect, url_for, render_template, Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import exc
from flask_whooshee import Whooshee
from datetime import datetime

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///database.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

db = SQLAlchemy(app)
whooshee = Whooshee(app)

@whooshee.register_model('topic_name')
class Topic(db.Model):
    __tablename__ = 'topic'
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    topic_name = db.Column(db.String, unique=True, nullable=False)
    topic_created = db.Column(db.DateTime, default=datetime.utcnow)
    topic_changed = db.Column(db.DateTime, default=datetime.utcnow)
    topic_tabs = db.relationship('Tab', backref='topic', cascade='all, delete-orphan', lazy='dynamic')

@whooshee.register_model('tab_name', 'tab_link', 'tab_description')
class Tab(db.Model): …

我陷入了这个数据库搜索问题：我有一个带有不同复选框的 PyQt5 小部件。它们中的每一个都代表我数据库中的一列。例如，“worker_name”和“department”。复选框放置在行编辑附近，因此用户可以在该行编辑中插入一些文本，然后单击复选框和“查找”按钮。

比方说，用户在 lineEdit 中插入“John”，然后按“worker_name”复选框并通过以下方式获取该工人的所有文档

session.query(Doc).filter_by('worker_name'=text_from_lineEdit)

如果用户在两个行编辑中插入文本并按下两个复选框（“worker_name”和“department”），查询将如下所示：

session.query(Doc).filter_by('worker_name'=text_from_lineEdit, 'department'=text_from_lineEdit_2)

但是，如果我有很多复选框（与数据库表中的列一样多）并且我不知道其中哪些会被按下，哪些不会被按下，该怎么办？在这种情况下我应该如何形成查询？我的意思是我将有“wrk_name”、“department”和“date”，用户只能通过“worker_name”或“worker_name”+“date”或所有复选框进行搜索。

有没有什么方法可以根据给出的参数以及不将它们包含在“filter_by”/“filter 中或只是忽略来动态地形成数据库查询。

请分享您对此类功能的可能实现的想法。谢谢您的时间。