diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index b96104ccae..cb1082e864 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -40,6 +40,9 @@ import synapse.metrics
from synapse.events import EventBase # noqa: F401
from synapse.events.snapshot import EventContext # noqa: F401
+from six.moves import range
+from six import itervalues, iteritems
+
from prometheus_client import Counter
logger = logging.getLogger(__name__)
@@ -245,7 +248,7 @@ class EventsStore(EventsWorkerStore):
partitioned.setdefault(event.room_id, []).append((event, ctx))
deferreds = []
- for room_id, evs_ctxs in partitioned.iteritems():
+ for room_id, evs_ctxs in iteritems(partitioned):
d = self._event_persist_queue.add_to_queue(
room_id, evs_ctxs,
backfilled=backfilled,
@@ -330,7 +333,7 @@ class EventsStore(EventsWorkerStore):
chunks = [
events_and_contexts[x:x + 100]
- for x in xrange(0, len(events_and_contexts), 100)
+ for x in range(0, len(events_and_contexts), 100)
]
for chunk in chunks:
@@ -364,7 +367,7 @@ class EventsStore(EventsWorkerStore):
(event, context)
)
- for room_id, ev_ctx_rm in events_by_room.iteritems():
+ for room_id, ev_ctx_rm in iteritems(events_by_room):
# Work out new extremities by recursively adding and removing
# the new events.
latest_event_ids = yield self.get_latest_event_ids_in_room(
@@ -459,12 +462,12 @@ class EventsStore(EventsWorkerStore):
event_counter.labels(event.type, origin_type, origin_entity).inc()
- for room_id, new_state in current_state_for_room.iteritems():
+ for room_id, new_state in iteritems(current_state_for_room):
self.get_current_state_ids.prefill(
(room_id, ), new_state
)
- for room_id, latest_event_ids in new_forward_extremeties.iteritems():
+ for room_id, latest_event_ids in iteritems(new_forward_extremeties):
self.get_latest_event_ids_in_room.prefill(
(room_id,), list(latest_event_ids)
)
@@ -641,20 +644,20 @@ class EventsStore(EventsWorkerStore):
"""
existing_state = yield self.get_current_state_ids(room_id)
- existing_events = set(existing_state.itervalues())
- new_events = set(ev_id for ev_id in current_state.itervalues())
+ existing_events = set(itervalues(existing_state))
+ new_events = set(ev_id for ev_id in itervalues(current_state))
changed_events = existing_events ^ new_events
if not changed_events:
return
to_delete = {
- key: ev_id for key, ev_id in existing_state.iteritems()
+ key: ev_id for key, ev_id in iteritems(existing_state)
if ev_id in changed_events
}
events_to_insert = (new_events - existing_events)
to_insert = {
- key: ev_id for key, ev_id in current_state.iteritems()
+ key: ev_id for key, ev_id in iteritems(current_state)
if ev_id in events_to_insert
}
@@ -757,11 +760,11 @@ class EventsStore(EventsWorkerStore):
)
def _update_current_state_txn(self, txn, state_delta_by_room, max_stream_order):
- for room_id, current_state_tuple in state_delta_by_room.iteritems():
+ for room_id, current_state_tuple in iteritems(state_delta_by_room):
to_delete, to_insert = current_state_tuple
txn.executemany(
"DELETE FROM current_state_events WHERE event_id = ?",
- [(ev_id,) for ev_id in to_delete.itervalues()],
+ [(ev_id,) for ev_id in itervalues(to_delete)],
)
self._simple_insert_many_txn(
@@ -774,7 +777,7 @@ class EventsStore(EventsWorkerStore):
"type": key[0],
"state_key": key[1],
}
- for key, ev_id in to_insert.iteritems()
+ for key, ev_id in iteritems(to_insert)
],
)
@@ -793,7 +796,7 @@ class EventsStore(EventsWorkerStore):
"event_id": ev_id,
"prev_event_id": to_delete.get(key, None),
}
- for key, ev_id in state_deltas.iteritems()
+ for key, ev_id in iteritems(state_deltas)
]
)
@@ -836,7 +839,7 @@ class EventsStore(EventsWorkerStore):
def _update_forward_extremities_txn(self, txn, new_forward_extremities,
max_stream_order):
- for room_id, new_extrem in new_forward_extremities.iteritems():
+ for room_id, new_extrem in iteritems(new_forward_extremities):
self._simple_delete_txn(
txn,
table="event_forward_extremities",
@@ -854,7 +857,7 @@ class EventsStore(EventsWorkerStore):
"event_id": ev_id,
"room_id": room_id,
}
- for room_id, new_extrem in new_forward_extremities.iteritems()
+ for room_id, new_extrem in iteritems(new_forward_extremities)
for ev_id in new_extrem
],
)
@@ -871,7 +874,7 @@ class EventsStore(EventsWorkerStore):
"event_id": event_id,
"stream_ordering": max_stream_order,
}
- for room_id, new_extrem in new_forward_extremities.iteritems()
+ for room_id, new_extrem in iteritems(new_forward_extremities)
for event_id in new_extrem
]
)
@@ -899,7 +902,7 @@ class EventsStore(EventsWorkerStore):
new_events_and_contexts[event.event_id] = (event, context)
else:
new_events_and_contexts[event.event_id] = (event, context)
- return new_events_and_contexts.values()
+ return list(new_events_and_contexts.values())
def _update_room_depths_txn(self, txn, events_and_contexts, backfilled):
"""Update min_depth for each room
@@ -925,7 +928,7 @@ class EventsStore(EventsWorkerStore):
event.depth, depth_updates.get(event.room_id, event.depth)
)
- for room_id, depth in depth_updates.iteritems():
+ for room_id, depth in iteritems(depth_updates):
self._update_min_depth_for_room_txn(txn, room_id, depth)
def _update_outliers_txn(self, txn, events_and_contexts):
@@ -1309,7 +1312,7 @@ class EventsStore(EventsWorkerStore):
" WHERE e.event_id IN (%s)"
) % (",".join(["?"] * len(ev_map)),)
- txn.execute(sql, ev_map.keys())
+ txn.execute(sql, list(ev_map))
rows = self.cursor_to_dict(txn)
for row in rows:
event = ev_map[row["event_id"]]
@@ -1572,7 +1575,7 @@ class EventsStore(EventsWorkerStore):
chunks = [
event_ids[i:i + 100]
- for i in xrange(0, len(event_ids), 100)
+ for i in range(0, len(event_ids), 100)
]
for chunk in chunks:
ev_rows = self._simple_select_many_txn(
@@ -1986,7 +1989,7 @@ class EventsStore(EventsWorkerStore):
logger.info("[purge] finding state groups which depend on redundant"
" state groups")
remaining_state_groups = []
- for i in xrange(0, len(state_rows), 100):
+ for i in range(0, len(state_rows), 100):
chunk = [sg for sg, in state_rows[i:i + 100]]
# look for state groups whose prev_state_group is one we are about
# to delete
@@ -2042,7 +2045,7 @@ class EventsStore(EventsWorkerStore):
"state_key": key[1],
"event_id": state_id,
}
- for key, state_id in curr_state.iteritems()
+ for key, state_id in iteritems(curr_state)
],
)
diff --git a/synapse/storage/presence.py b/synapse/storage/presence.py
index 9e9d3c2591..f05d91cc58 100644
--- a/synapse/storage/presence.py
+++ b/synapse/storage/presence.py
@@ -16,6 +16,7 @@
from ._base import SQLBaseStore
from synapse.api.constants import PresenceState
from synapse.util.caches.descriptors import cached, cachedInlineCallbacks, cachedList
+from synapse.util import batch_iter
from collections import namedtuple
from twisted.internet import defer
@@ -115,11 +116,7 @@ class PresenceStore(SQLBaseStore):
" AND user_id IN (%s)"
)
- batches = (
- presence_states[i:i + 50]
- for i in xrange(0, len(presence_states), 50)
- )
- for states in batches:
+ for states in batch_iter(presence_states, 50):
args = [stream_id]
args.extend(s.user_id for s in states)
txn.execute(
diff --git a/synapse/storage/search.py b/synapse/storage/search.py
index a9c299a861..f0fa5d7631 100644
--- a/synapse/storage/search.py
+++ b/synapse/storage/search.py
@@ -448,7 +448,7 @@ class SearchStore(BackgroundUpdateStore):
"search_msgs", self.cursor_to_dict, sql, *args
)
- results = filter(lambda row: row["room_id"] in room_ids, results)
+ results = list(filter(lambda row: row["room_id"] in room_ids, results))
events = yield self._get_events([r["event_id"] for r in results])
@@ -603,7 +603,7 @@ class SearchStore(BackgroundUpdateStore):
"search_rooms", self.cursor_to_dict, sql, *args
)
- results = filter(lambda row: row["room_id"] in room_ids, results)
+ results = list(filter(lambda row: row["room_id"] in room_ids, results))
events = yield self._get_events([r["event_id"] for r in results])
diff --git a/synapse/storage/state.py b/synapse/storage/state.py
index ffa4246031..bdee14a8eb 100644
--- a/synapse/storage/state.py
+++ b/synapse/storage/state.py
@@ -16,6 +16,9 @@
from collections import namedtuple
import logging
+from six import iteritems, itervalues
+from six.moves import range
+
from twisted.internet import defer
from synapse.storage.background_updates import BackgroundUpdateStore
@@ -134,7 +137,7 @@ class StateGroupWorkerStore(SQLBaseStore):
event_ids,
)
- groups = set(event_to_groups.itervalues())
+ groups = set(itervalues(event_to_groups))
group_to_state = yield self._get_state_for_groups(groups)
defer.returnValue(group_to_state)
@@ -166,18 +169,18 @@ class StateGroupWorkerStore(SQLBaseStore):
state_event_map = yield self.get_events(
[
- ev_id for group_ids in group_to_ids.itervalues()
- for ev_id in group_ids.itervalues()
+ ev_id for group_ids in itervalues(group_to_ids)
+ for ev_id in itervalues(group_ids)
],
get_prev_content=False
)
defer.returnValue({
group: [
- state_event_map[v] for v in event_id_map.itervalues()
+ state_event_map[v] for v in itervalues(event_id_map)
if v in state_event_map
]
- for group, event_id_map in group_to_ids.iteritems()
+ for group, event_id_map in iteritems(group_to_ids)
})
@defer.inlineCallbacks
@@ -186,7 +189,7 @@ class StateGroupWorkerStore(SQLBaseStore):
"""
results = {}
- chunks = [groups[i:i + 100] for i in xrange(0, len(groups), 100)]
+ chunks = [groups[i:i + 100] for i in range(0, len(groups), 100)]
for chunk in chunks:
res = yield self.runInteraction(
"_get_state_groups_from_groups",
@@ -347,21 +350,21 @@ class StateGroupWorkerStore(SQLBaseStore):
event_ids,
)
- groups = set(event_to_groups.itervalues())
+ groups = set(itervalues(event_to_groups))
group_to_state = yield self._get_state_for_groups(groups, types)
state_event_map = yield self.get_events(
- [ev_id for sd in group_to_state.itervalues() for ev_id in sd.itervalues()],
+ [ev_id for sd in itervalues(group_to_state) for ev_id in itervalues(sd)],
get_prev_content=False
)
event_to_state = {
event_id: {
k: state_event_map[v]
- for k, v in group_to_state[group].iteritems()
+ for k, v in iteritems(group_to_state[group])
if v in state_event_map
}
- for event_id, group in event_to_groups.iteritems()
+ for event_id, group in iteritems(event_to_groups)
}
defer.returnValue({event: event_to_state[event] for event in event_ids})
@@ -384,12 +387,12 @@ class StateGroupWorkerStore(SQLBaseStore):
event_ids,
)
- groups = set(event_to_groups.itervalues())
+ groups = set(itervalues(event_to_groups))
group_to_state = yield self._get_state_for_groups(groups, types)
event_to_state = {
event_id: group_to_state[group]
- for event_id, group in event_to_groups.iteritems()
+ for event_id, group in iteritems(event_to_groups)
}
defer.returnValue({event: event_to_state[event] for event in event_ids})
@@ -503,7 +506,7 @@ class StateGroupWorkerStore(SQLBaseStore):
got_all = is_all or not missing_types
return {
- k: v for k, v in state_dict_ids.iteritems()
+ k: v for k, v in iteritems(state_dict_ids)
if include(k[0], k[1])
}, missing_types, got_all
@@ -562,12 +565,12 @@ class StateGroupWorkerStore(SQLBaseStore):
# Now we want to update the cache with all the things we fetched
# from the database.
- for group, group_state_dict in group_to_state_dict.iteritems():
+ for group, group_state_dict in iteritems(group_to_state_dict):
state_dict = results[group]
state_dict.update(
((intern_string(k[0]), intern_string(k[1])), to_ascii(v))
- for k, v in group_state_dict.iteritems()
+ for k, v in iteritems(group_state_dict)
)
self._state_group_cache.update(
@@ -654,7 +657,7 @@ class StateGroupWorkerStore(SQLBaseStore):
"state_key": key[1],
"event_id": state_id,
}
- for key, state_id in delta_ids.iteritems()
+ for key, state_id in iteritems(delta_ids)
],
)
else:
@@ -669,7 +672,7 @@ class StateGroupWorkerStore(SQLBaseStore):
"state_key": key[1],
"event_id": state_id,
}
- for key, state_id in current_state_ids.iteritems()
+ for key, state_id in iteritems(current_state_ids)
],
)
@@ -794,11 +797,11 @@ class StateStore(StateGroupWorkerStore, BackgroundUpdateStore):
"state_group": state_group_id,
"event_id": event_id,
}
- for event_id, state_group_id in state_groups.iteritems()
+ for event_id, state_group_id in iteritems(state_groups)
],
)
- for event_id, state_group_id in state_groups.iteritems():
+ for event_id, state_group_id in iteritems(state_groups):
txn.call_after(
self._get_state_group_for_event.prefill,
(event_id,), state_group_id
@@ -826,7 +829,7 @@ class StateStore(StateGroupWorkerStore, BackgroundUpdateStore):
def reindex_txn(txn):
new_last_state_group = last_state_group
- for count in xrange(batch_size):
+ for count in range(batch_size):
txn.execute(
"SELECT id, room_id FROM state_groups"
" WHERE ? < id AND id <= ?"
@@ -884,7 +887,7 @@ class StateStore(StateGroupWorkerStore, BackgroundUpdateStore):
# of keys
delta_state = {
- key: value for key, value in curr_state.iteritems()
+ key: value for key, value in iteritems(curr_state)
if prev_state.get(key, None) != value
}
@@ -924,7 +927,7 @@ class StateStore(StateGroupWorkerStore, BackgroundUpdateStore):
"state_key": key[1],
"event_id": state_id,
}
- for key, state_id in delta_state.iteritems()
+ for key, state_id in iteritems(delta_state)
],
)
diff --git a/synapse/storage/user_directory.py b/synapse/storage/user_directory.py
index d6e289ffbe..275c299998 100644
--- a/synapse/storage/user_directory.py
+++ b/synapse/storage/user_directory.py
@@ -22,6 +22,8 @@ from synapse.api.constants import EventTypes, JoinRules
from synapse.storage.engines import PostgresEngine, Sqlite3Engine
from synapse.types import get_domain_from_id, get_localpart_from_id
+from six import iteritems
+
import re
import logging
@@ -100,7 +102,7 @@ class UserDirectoryStore(SQLBaseStore):
user_id, get_localpart_from_id(user_id), get_domain_from_id(user_id),
profile.display_name,
)
- for user_id, profile in users_with_profile.iteritems()
+ for user_id, profile in iteritems(users_with_profile)
)
elif isinstance(self.database_engine, Sqlite3Engine):
sql = """
@@ -112,7 +114,7 @@ class UserDirectoryStore(SQLBaseStore):
user_id,
"%s %s" % (user_id, p.display_name,) if p.display_name else user_id
)
- for user_id, p in users_with_profile.iteritems()
+ for user_id, p in iteritems(users_with_profile)
)
else:
# This should be unreachable.
@@ -130,7 +132,7 @@ class UserDirectoryStore(SQLBaseStore):
"display_name": profile.display_name,
"avatar_url": profile.avatar_url,
}
- for user_id, profile in users_with_profile.iteritems()
+ for user_id, profile in iteritems(users_with_profile)
]
)
for user_id in users_with_profile:
|
