Sqlalchemy ، تحويل قائمة المعرفات إلى قائمة الكائنات

Question

لدي تسلسل من IDS أريد استرداده. انه سهل:

session.query(Record).filter(Record.id.in_(seq)).all()

هل هناك طريقة افضل لفعله؟

Answer 1

الرمز الخاص بك غرامة طيبة.

IN مثل حفنة من X=Y انضم إليها OR وسريع جدا في قواعد البيانات المعاصرة.

ومع ذلك ، إذا كانت قائمة المعرفات الخاصة بك طويلة ، فيمكنك جعل الاستعلام أكثر كفاءة قليلاً من خلال تمرير مسابقة فرعية لإرجاع قائمة المعرفات.

Answer 2

الرمز كما هو جيد تماما. ومع ذلك ، يسألني أحدهم عن بعض نظام التحوط بين النهجين لقيامهما كبيرًا مقابل استخدام GET () للمعرفات الفردية.

إذا كان شخص ما يحاول حقًا تجنب الاختيار ، فإن أفضل طريقة للقيام بذلك هي إعداد الكائنات التي تحتاجها في الذاكرة في وقت مبكر. مثل ، أنت تعمل على جدول كبير من العناصر. قم بتقسيم العمل إلى أجزاء ، مثل ، اطلب مجموعة العمل الكاملة حسب المفتاح الأساسي ، أو عن طريق التاريخ ، أيا كان ، ثم قم بتحميل كل شيء لهذا الجزء محليًا إلى ذاكرة التخزين المؤقت:

 all_ids = [<huge list of ids>]

 all_ids.sort()
 while all_ids:
     chunk = all_ids[0:1000]

     # bonus exercise!  Throw each chunk into a multiprocessing.pool()!
     all_ids = all_ids[1000:]

     my_cache = dict(
           Session.query(Record.id, Record).filter(
                 Record.id.between(chunk[0], chunk[-1]))
     )

     for id_ in chunk:
         my_obj = my_cache[id_]
         <work on my_obj>

هذا هو قضية استخدام العالم الحقيقي.

ولكن لتوضيح بعض واجهة برمجة تطبيقات Sqlalchemy ، يمكننا إنشاء وظيفة تقوم بها السجلات التي لا نملكها وحصلت على من نقوم بها. هذا هو:

from sqlalchemy import inspect


def get_all(session, cls, seq):
    mapper = inspect(cls)
    lookup = set()
    for ident in seq:
        key = mapper.identity_key_from_primary_key((ident, ))
        if key in session.identity_map:
            yield session.identity_map[key]
        else:
            lookup.add(ident)
    if lookup:
        for obj in session.query(cls).filter(cls.id.in_(lookup)):
            yield obj

هنا مظاهرة:

from sqlalchemy import Column, Integer, create_engine, String
from sqlalchemy.orm import Session
from sqlalchemy.ext.declarative import declarative_base
import random

Base = declarative_base()


class A(Base):
    __tablename__ = 'a'
    id = Column(Integer, primary_key=True)
    data = Column(String)

e = create_engine("sqlite://", echo=True)
Base.metadata.create_all(e)

ids = range(1, 50)

s = Session(e)
s.add_all([A(id=i, data='a%d' % i) for i in ids])
s.commit()
s.close()

already_loaded = s.query(A).filter(A.id.in_(random.sample(ids, 10))).all()

assert len(s.identity_map) == 10

to_load = set(random.sample(ids, 25))
all_ = list(get_all(s, A, to_load))

assert set(x.id for x in all_) == to_load

Answer 3

إذا كنت تستخدم المفاتيح الأساسية المركبة ، فيمكنك استخدامها tuple_, ، كما في

from sqlalchemy import tuple_
session.query(Record).filter(tuple_(Record.id1, Record.id2).in_(seq)).all()

لاحظ أن هذا غير متوفر على SQLite (انظر مستند).

Answer 4

أوصي بإلقاء نظرة على SQL التي تنتجها. يمكنك فقط طباعة Str (استعلام) لرؤيته.

لست على علم بالطريقة المثالية للقيام بذلك باستخدام SQL القياسي.

Answer 5

هناك طريقة واحدة. إذا كان من المعقول توقع أن يتم بالفعل تحميل الكائنات المعنية في الجلسة ؛ لقد وصلت إليها من قبل في نفس المعاملة ، يمكنك بدلاً من ذلك القيام بها:

map(session.query(Record).get, seq)

في حالة وجود هذه الكائنات بالفعل ، سيكون هذا أسرع بكثير ، حيث لن يكون هناك أي استعلامات لاسترداد تلك الكائنات ؛ من ناحية أخرى ، إذا كان هناك أكثر من عدد صغير من هذه الكائنات ليس تم تحميله ، سيكون أبطأ بكثير ، لأنه سيؤدي إلى استفسار لكل مثيل مفقود ، بدلاً من استعلام واحد لجميع الكائنات.

يمكن أن يكون هذا مفيدًا عندما تفعل joinedload() الاستعلامات قبل الوصول إلى الخطوة أعلاه ، بحيث يمكنك التأكد من تحميلها بالفعل. بشكل عام ، يجب عليك استخدام الحل في السؤال افتراضيًا ، واستكشاف هذا الحل فقط عندما ترى أنك تستفسر عن نفس الكائنات مرارًا وتكرارًا.