相关疑难解决方法(0)

我有一个记录,我希望它存在于数据库中,如果它不存在,如果它已存在(主键存在)我希望字段更新到当前状态.这通常称为upsert.

以下不完整的代码片段演示了什么可行,但它似乎过于笨重(特别是如果有更多的列).什么是更好/最好的方式？

Base = declarative_base()
class Template(Base):
    __tablename__ = 'templates'
    id = Column(Integer, primary_key = True)
    name = Column(String(80), unique = True, index = True)
    template = Column(String(80), unique = True)
    description = Column(String(200))
    def __init__(self, Name, Template, Desc):
        self.name = Name
        self.template = Template
        self.description = Desc

def UpsertDefaultTemplate():
    sess = Session()
    desired_default = Template("default", "AABBCC", "This is the default template")
    try:
        q = sess.query(Template).filter_by(name = desiredDefault.name)
        existing_default = q.one()
    except sqlalchemy.orm.exc.NoResultFound:
        #default does not exist yet, …

INSERT ... ON DUPLICATE KEY UPDATE在SQLAlchemy中有一种优雅的方式吗？我的意思是语法类似的东西inserter.insert().execute(list_of_dictionaries)？

这是一个功能.我的意图是使用关键字参数默认值使字典成为空字典(如果未提供).

>>> def f( i, d={}, x=3 ) :
...     d[i] = i*i
...     x += i
...     return x, d
... 
>>> f( 2 )
(5, {2: 4})

但是当我接下来打电话给f时,我得到:

>>> f(3)
(6, {2: 4, 3: 9})

看起来第二次调用时的关键字参数d不指向空字典,而是指向前一次调用结束时留下的字典.每次通话时,号码x重置为3.

现在我可以解决这个问题,但我希望你的帮助能够理解这一点.我相信关键字参数在函数的本地范围内,并且一旦函数返回就会被删除.(如果我不精确,请原谅并纠正我的术语.)

因此,应删除名称d指向的本地值,并在下一次调用时,如果我不提供关键字参数d,则应将d设置为default {}.但正如您所看到的,d被设置为前一个调用中指向的字典.

到底是怎么回事？

是literal {}在封闭范围的防线？

在2.5,2.6和3.1中可以看到此行为.

我正在寻找创建新SQLAlchemy对象的最快方法,只有在数据库中不存在它时.

我现在这样做的方法是首先获取查询的计数以查看它是否存在,如果不存在 - 然后我创建它.例如:

if not User.query.filter(email=user.email).count():
    db.session.add(user)
    db.session.commit()

这是最好的方法吗？会喜欢一些反馈.谢谢!

我从多个进程插入一个SQL数据库.过程有时可能会尝试将重复数据插入表中.我试图以一种处理重复的方式编写查询,但我仍然得到:

System.Data.SqlClient.SqlException: Violation of UNIQUE KEY constraint 'UK1_MyTable'. Cannot insert duplicate key in object 'dbo.MyTable'.
The statement has been terminated.

我的查询看起来像:

INSERT INTO MyTable (FieldA, FieldB, FieldC)
SELECT FieldA='AValue', FieldB='BValue', FieldC='CValue'
WHERE (SELECT COUNT(*) FROM MyTable WHERE FieldA='AValue' AND FieldB='BValue' AND FieldC='CValue' ) = 0

约束'UK1_MyConstraint'表示在MyTable中,3个字段的组合应该是唯一的.

我的问题:

1. 为什么这不起作用？
2. 我需要进行哪些修改才能因违反约束而无法出现异常？

请注意,我知道还有其他方法可以解决"INSERT if not exists"的原始问题,例如(摘要):

• 使用TRY CATCH
• 如果不存在INSERT(在具有可序列化隔离的事务中)

我应该使用其中一种方法吗？

编辑1个 SQL用于创建表:

CREATE TABLE [dbo].[MyTable](
  [Id] [bigint] IDENTITY(1,1) NOT NULL,
  [FieldA] [bigint] NOT NULL,
  [FieldB] [int] NOT NULL,
  [FieldC] [char](3) NULL,
  [FieldD] [float] …

我们有 2 个表定义如下

CREATE TABLE foo (
  id BIGSERIAL PRIMARY KEY,
  name TEXT NOT NULL UNIQUE
);

CREATE TABLE bar (
  foo_id BIGINT UNIQUE,
  foo_name TEXT NOT NULL UNIQUE REFERENCES foo (name)
);

我注意到在同时执行以下两个查询时

INSERT INTO foo (name) VALUES ('BAZ')

INSERT INTO bar (foo_name, foo_id) VALUES ('BAZ', (SELECT id FROM foo WHERE name = 'BAZ'))

在某些情况下，最终可能会在barwhere foo_idis 中插入一行NULL。这两个查询由两个完全不同的进程在不同的事务中执行。

这怎么可能？我希望第二条语句要么由于外键违规而失败（如果记录foo不存在），要么以非空值foo_id（如果是）成功。

是什么导致了这种竞争条件？是由于子选择，还是由于检查外键约束的时间？

我们使用隔离级别“已提交读”和 postgres 10.3 版。

编辑

我认为这个问题并不清楚是什么让我感到困惑。问题是关于在执行单个语句期间如何以及为什么观察到数据库的 2 个不同状态。subselect 观察到 foo 中的记录不存在，而 …

我有一个类,Item它id是一个主键并自动生成.现在我从其他外部源读取数据,创建一个Item对象,并需要检查我的items表中是否已存在此对象.我该怎么做？

我正在使用SQLAlchemy(和Elixir)用SQLite作为数据库后端在Python中编写应用程序.我使用代码启动一个新的事务session.begin_transaction(),但是当我调用时,session.rollback()我收到以下错误:

sqlalchemy.exceptions.OperationalError: (OperationalError) no such savepoint: sa_savepoint_1 u'ROLLBACK TO SAVEPOINT sa_savepoint_1' []

我也收到类似的错误session.commit().据我所知,sqlite支持SAVEPOINTS(http://www.sqlite.org/lang_savepoint.html).

如何使嵌套事务生效？

我正在使用类似于此问题的答案的 get_or_create 模式：

SQLAlchemy 是否有相当于 Django 的 get_or_create 的功能？

但是我在另一个线程在第一个线程的“选择”和“插入”之间创建实例（具有相同的 pk）时遇到问题。

get_or_create 函数是否应该锁定两个查询的表？使用 SqlAlchemy 实现锁定的最佳方法是什么？

我正在使用 postgresql：http ://www.postgresql.org/docs/current/static/sql-lock.html

我能看到的唯一 sqlalchemy 锁定功能是“ for update ”，这似乎不是正确的锁定类型？

我是SQLAlchemy的新手,并阅读了基本文档.我目前正在关注Mike Driscoll的MediaLocker教程并根据自己的目的修改/扩展它.

我有三张桌子(贷款,人,卡).贷款卡和贷款人都是一对多的关系,并建模如下:

from sqlalchemy import Table, Column, DateTime, Integer, ForeignKey, Unicode
from sqlalchemy.orm import backref, relation
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base

engine = create_engine("sqlite:///cardsys.db", echo=True)
DeclarativeBase = declarative_base(engine)
metadata = DeclarativeBase.metadata

class Loan(DeclarativeBase):
    """
    Loan model
    """

    __tablename__ = "loans"

    id = Column(Integer, primary_key=True)
    card_id = Column(Unicode, ForeignKey("cards.id"))
    person_id = Column(Unicode, ForeignKey("people.id"))
    date_issued = Column(DateTime)
    date_due = Column(DateTime)
    date_returned = Column(DateTime)
    issue_reason = Column(Unicode(50))
    person = relation("Person", backref="loans", cascade_backrefs=False)
    card = relation("Card", backref="loans", cascade_backrefs=False) …

所以我对SQLAlchemy很新.

我有一个模型显示表中有大约10,000行.这是班级:

class Showing(Base):
    __tablename__   = "showings"

    id              = Column(Integer, primary_key=True)
    time            = Column(DateTime)
    link            = Column(String)
    film_id         = Column(Integer, ForeignKey('films.id'))
    cinema_id       = Column(Integer, ForeignKey('cinemas.id'))

    def __eq__(self, other):
        if self.time == other.time and self.cinema == other.cinema and self.film == other.film:
            return True
        else:
            return False

任何人都可以给我一些关于插入新表现的最快方法的指导,如果它还不存在的话.我觉得它稍微复杂一点,因为如果时间,电影和电影在表演中是独一无二的,那么表演才是唯一的.

我目前有这个代码:

def AddShowings(self, showing_times, cinema, film):
    all_showings = self.session.query(Showing).options(joinedload(Showing.cinema), joinedload(Showing.film)).all()
    for showing_time in showing_times:
        tmp_showing = Showing(time=showing_time[0], film=film, cinema=cinema, link=showing_time[1])
        if tmp_showing not in all_showings:
            self.session.add(tmp_showing)
            self.session.commit()
            all_showings.append(tmp_showing)

哪个有效,但似乎很慢.任何帮助深表感谢.

我有一个关于如何检查条目是否已经在数据库表中的更合适/更有效的方法的问题。

假设我有一个Foo用单个属性调用的表，Name它也是unique。

如果已经存在具有此类名称的条目，在插入之前检查的更好方法是什么？我可以想到两种选择：

from sqlalchemy import MetaData, Table, Column, String, \
create_engine
from sqlalchemy.orm import mapper, relationship, sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.exc import IntegrityError

Base = declarative_base()

class Foo(Base):
    __tablename__ = "foo"

    name = Column(String(), primary_key=True)

    @staticmethod
    def silent_insert1(name):
            if not session.query(Foo).filter(Foo.name == name).first():
                    foo = Foo()
                    foo.name = name
                    session.add(foo)
                    session.commit()
            else:
                    print("already exists 1")

    @staticmethod
    def silent_insert2(name):
            try:
                    foo = Foo()
                    foo.name = name
                    session.add(foo)
                    session.commit()
            except(IntegrityError):
                    print("already exists 2") …

所有,

我正在读一个csv文件并使用sqlalchemy将数据添加到MySQL数据库.其中一个表是地址表,它只能保存唯一的地址.这些地址与另一个具有地址id的外键字段的"语句"表之间存在关系.

因此,对于我的数据文件中的每一行,我创建一个新的语句obj,然后获取相关地址的id.如果地址已存在,则返回该id.否则,我创建一个新的地址obj并返回该id.这是使用下面的代码完成的,改编自这个SO问题.

def get_or_create(self, model, rec):
    instance = self.session.query(model).filter_by(**dict(filter(lambda (x,y): x in model.__dict__.keys(), rec.iteritems()))).first()
    if instance:
        return instance
    else:
        instance = model(rec)
        return instance

我正在使用GUID作为我的id字段,它是地址表的主键的一部分:

class address(Base):
    __tablename__ = 'address'
    id = id_column()
    name               = Column(String(75), primary_key=True)
    Address_Line_One   = Column(String(50), primary_key=True)
    Address_Line_Two   = Column(String(50), primary_key=True)
    Address_Line_Three = Column(String(50), primary_key=True)
    Address_Line_Four  = Column(String(50), primary_key=True)

将id_column()来自这里,虽然它已被转换为CHAR(32)由于限制别处.最后,这里有一个片段:

currStatement   = statements(rec, id=currGUID)
currStatement.address = self.get_or_create(address, rec)

这一切都很好,除非它很慢.对于在一个事务中插入的~65,000个语句,我看到干净测试DB上的1.5小时插入时间.实时观察插入显示它快速达到~10,000行,然后插入速度开始下降.

我该怎么做才能加快插入时间？

编辑:

经过进一步测试,我发现插入时间慢是因为每个对象都是单独插入的.所以,我有~65,000行,每行都有几个sqlalchemy对象,单独插入.使用sqlalchemy 0.7,我如何批量插入对象？