Блог

У меня есть некоторые данные временных рядов, где у меня есть наборы временных рядов, каждый Timeseries экземпляр которых имеет отношение «один ко многим» с Point экземплярами. Ниже приведено упрощенное представление данных.

tables.py:

 class Timeseries(Base):
    __tablename__ = "timeseries"

    id = Column("id", Integer, primary_key=True)
    points = relationship("Point", back_populates="ts")


class Point(Base):
    __tablename__ = "point"

    id = Column("id", Integer, primary_key=True)
    t = Column("t", Float)
    v = Column("v", Float)
    ts_id = Column(Integer, ForeignKey("timeseries.id"))
    ts = relationship("Timeseries", back_populates="points")
  

Вопрос: Я пытаюсь составить запрос с такими столбцами: «timeseries_id», «id», «t», «v», «id_next», «t_next», «v_next». То есть я хочу иметь возможность видеть данные каждой точки рядом с данными следующих точек во временном ряду в хронологическом порядке, но я изо всех сил пытаюсь получить таблицу, в которой нет элементов из неявного объединения? (Редактировать: Важным моментом является то, что я хочу иметь возможность получать этот список, используя 100% запросов и объектов подзапроса в sqlalchemy, потому что мне нужно использовать эту запрашиваемую таблицу в дальнейших объединениях, фильтрах и т.д.) Вот основное начало того, что я получил, (обратите внимание, что я не запускал этот код, поскольку это упрощенная версия моей реальной базы данных, но идея та же):

 # The point data actually in the database.
sq = (session.query(
    Timeseries.id.label("timeseries_id"),
    Point.id,
    Point.t,
    Point.v)
.select_from(
    join(Timeseries, Point, Timeseries.id==Point.ts_id))
.group_by('timeseries_id')
.subquery())

# first point manually added to each list in query
sq_first = (session.query(
    Timeseries.id.label("timeseries_id"),
    sa.literal_column("-1", Integer).label("id"), # Some unused Point.id value
    sa.literal_column(-math.inf, Float).label("t"),
    sa.literal_column(-math.inf, Float).label("v"))
.select_from(
    join(Timeseries, Point, Timeseries.id==Point.ts_id))
.subquery())

# last point manually added to each list in query.
sq_last = (session.query(
    Timeseries.id.label("timeseries_id"),
    sa.literal_column("-2", Integer).label("id"), # Another unused Point.id value
    sa.literal_column(math.inf, Float).label("t"),
    sa.literal_column(math.inf, Float).label("v"))
.select_from(
    join(Timeseries, Point, Timeseries.id==Point.ts_id))
.subquery())

# Append each timeseries in `sq` table with last point
sq_points_curr = session.query(sa.union_all(sq_first, sq)).subquery()
sq_points_next = session.query(sa.union_all(sq, sq_last)).subquery()
  

Предполагая, что то, что я сделал до сих пор, полезно, это та часть, где я застрял:

 #I guess rename the columns in `sq_points_next` to append them by "_next"....
sq_points_next = (session.query(
    sq_points_curr.c.timeseries_id
    sq_points_curr.c.id.label("id_next"),
    sq_points_curr.c.t.label("t_next"),
    sq_points_curr.c.v.label("v_next"))
.subquery())

# ... and then perform a join along "timeseries_id" somehow to get the table I originally wanted...
sq_point_pairs = (session.query(
    Timeseries.id.label("timeseries_id")
    "id",
    "t",
    "v",
    "id_next",
    "t_next",
    "v_next"
).select_from(
    sq_points, sq_points_next, sq_points.timeseries_id==sq_points_next.timeseries_id)
)
  

Я даже не уверен, будет ли это последнее скомпилировано на данном этапе, поскольку опять же оно адаптировано / упрощено из реального кода, но это не дает таблицу смежных моментов времени и т.д..

Редактировать (10 августа 2019):

Следующий упрощенный запрос от Nathan, безусловно, является правильным подходом, близким к рабочему, но вызывает ошибки для sqlite.

 sq = session.query(
        Timeseries.id.label("timeseries_id"),
        Point.t.label("point_t"),
        func.lead(Point.t).over().label('point_after_t')
    ).select_from(
        join(Timeseries, Point, Timeseries.id == Point.ts_id)
    ).order_by(Timeseries.id)

print(sq.all())
  

Предполагая, что вы можете получить достаточно свежую версию модуля sqlite3 python (например, с помощью Anaconda), вы можете использовать LEAD функцию window для достижения своей цели. Чтобы использовать результаты LEAD функции в дальнейших запросах, вам также нужно будет использовать CTE. Следующий подход сработал для меня со схемой, которую вы дали:

 sq = session.query(
        Timeseries.id.label("timeseries_id"),
        Point.id.label("point_id"),
        Point.t.label("point_t"),
        Point.v.label("point_v"),
        func.lead(Point.id).over().label('point_after_id'),
        func.lead(Point.v).over().label('point_after_v'),
        func.lead(Point.t).over().label('point_after_t')).select_from(
            join(Timeseries, Point, Timeseries.id == Point.ts_id)).order_by(Timeseries.id)

with_after = sq.cte()
session.execute(with_after.select().where(
        with_after.c.point_v < with_after.c.point_after_v)).fetchall()
  

Вместо того, чтобы прыгать через обручи, чтобы заставить запрос выдавать парные результаты, которые вы ищете, почему бы просто не извлечь все points данные, относящиеся к определенной Timeseries строке, а затем рекомбинировать данные в пары, которые вы ищете? Например:

 from operator import attrgetter

def to_dict(a, b):
    # formats a pair of points rows into a dict object
    return {
        'timeseries_id': a.ts_id,
        'id': a.id, 't': a.t, 'v': a.v,
        'id_next': b.id, 't_next': b.t, 'v_next': b.v
    }      

def timeseries_pairs(session, ts_id):
        # queries the db for particular Timeseries row, and combines points pairs
        ts = session.query(Timeseries).
            filter(Timeseries.id == ts_id).
            first()

        ts.points.sort(key=attrgetter('t'))
        pairs = [to_dict(a, b) for a, b in zip(ts.points, ts.points[1:])]
        last = ts.points[-1]
        pairs.append({
            'timeseries_id': last.ts_id,
            'id': last.id, 't': last.t, 'v': last.v,
            'id_next': None, 't_next': None, 'v_next': None
            })

        return pairs

# pass the session and a timeseries id to return a list of points pairs
timeseries_pairs(session, 1)