diff --git a/synapse/storage/appservice.py b/synapse/storage/appservice.py
index 31248d5e06..6092f600ba 100644
--- a/synapse/storage/appservice.py
+++ b/synapse/storage/appservice.py
@@ -51,8 +51,7 @@ def _make_exclusive_regex(services_cache):
class ApplicationServiceWorkerStore(SQLBaseStore):
def __init__(self, db_conn, hs):
self.services_cache = load_appservices(
- hs.hostname,
- hs.config.app_service_config_files
+ hs.hostname, hs.config.app_service_config_files
)
self.exclusive_user_regex = _make_exclusive_regex(self.services_cache)
@@ -122,8 +121,9 @@ class ApplicationServiceStore(ApplicationServiceWorkerStore):
pass
-class ApplicationServiceTransactionWorkerStore(ApplicationServiceWorkerStore,
- EventsWorkerStore):
+class ApplicationServiceTransactionWorkerStore(
+ ApplicationServiceWorkerStore, EventsWorkerStore
+):
@defer.inlineCallbacks
def get_appservices_by_state(self, state):
"""Get a list of application services based on their state.
@@ -135,9 +135,7 @@ class ApplicationServiceTransactionWorkerStore(ApplicationServiceWorkerStore,
may be empty.
"""
results = yield self._simple_select_list(
- "application_services_state",
- dict(state=state),
- ["as_id"]
+ "application_services_state", dict(state=state), ["as_id"]
)
# NB: This assumes this class is linked with ApplicationServiceStore
as_list = self.get_app_services()
@@ -180,9 +178,7 @@ class ApplicationServiceTransactionWorkerStore(ApplicationServiceWorkerStore,
A Deferred which resolves when the state was set successfully.
"""
return self._simple_upsert(
- "application_services_state",
- dict(as_id=service.id),
- dict(state=state)
+ "application_services_state", dict(as_id=service.id), dict(state=state)
)
def create_appservice_txn(self, service, events):
@@ -195,6 +191,7 @@ class ApplicationServiceTransactionWorkerStore(ApplicationServiceWorkerStore,
Returns:
AppServiceTransaction: A new transaction.
"""
+
def _create_appservice_txn(txn):
# work out new txn id (highest txn id for this service += 1)
# The highest id may be the last one sent (in which case it is last_txn)
@@ -204,7 +201,7 @@ class ApplicationServiceTransactionWorkerStore(ApplicationServiceWorkerStore,
txn.execute(
"SELECT MAX(txn_id) FROM application_services_txns WHERE as_id=?",
- (service.id,)
+ (service.id,),
)
highest_txn_id = txn.fetchone()[0]
if highest_txn_id is None:
@@ -217,16 +214,11 @@ class ApplicationServiceTransactionWorkerStore(ApplicationServiceWorkerStore,
txn.execute(
"INSERT INTO application_services_txns(as_id, txn_id, event_ids) "
"VALUES(?,?,?)",
- (service.id, new_txn_id, event_ids)
- )
- return AppServiceTransaction(
- service=service, id=new_txn_id, events=events
+ (service.id, new_txn_id, event_ids),
)
+ return AppServiceTransaction(service=service, id=new_txn_id, events=events)
- return self.runInteraction(
- "create_appservice_txn",
- _create_appservice_txn,
- )
+ return self.runInteraction("create_appservice_txn", _create_appservice_txn)
def complete_appservice_txn(self, txn_id, service):
"""Completes an application service transaction.
@@ -252,26 +244,26 @@ class ApplicationServiceTransactionWorkerStore(ApplicationServiceWorkerStore,
"appservice: Completing a transaction which has an ID > 1 from "
"the last ID sent to this AS. We've either dropped events or "
"sent it to the AS out of order. FIX ME. last_txn=%s "
- "completing_txn=%s service_id=%s", last_txn_id, txn_id,
- service.id
+ "completing_txn=%s service_id=%s",
+ last_txn_id,
+ txn_id,
+ service.id,
)
# Set current txn_id for AS to 'txn_id'
self._simple_upsert_txn(
- txn, "application_services_state", dict(as_id=service.id),
- dict(last_txn=txn_id)
+ txn,
+ "application_services_state",
+ dict(as_id=service.id),
+ dict(last_txn=txn_id),
)
# Delete txn
self._simple_delete_txn(
- txn, "application_services_txns",
- dict(txn_id=txn_id, as_id=service.id)
+ txn, "application_services_txns", dict(txn_id=txn_id, as_id=service.id)
)
- return self.runInteraction(
- "complete_appservice_txn",
- _complete_appservice_txn,
- )
+ return self.runInteraction("complete_appservice_txn", _complete_appservice_txn)
@defer.inlineCallbacks
def get_oldest_unsent_txn(self, service):
@@ -284,13 +276,14 @@ class ApplicationServiceTransactionWorkerStore(ApplicationServiceWorkerStore,
A Deferred which resolves to an AppServiceTransaction or
None.
"""
+
def _get_oldest_unsent_txn(txn):
# Monotonically increasing txn ids, so just select the smallest
# one in the txns table (we delete them when they are sent)
txn.execute(
"SELECT * FROM application_services_txns WHERE as_id=?"
" ORDER BY txn_id ASC LIMIT 1",
- (service.id,)
+ (service.id,),
)
rows = self.cursor_to_dict(txn)
if not rows:
@@ -301,8 +294,7 @@ class ApplicationServiceTransactionWorkerStore(ApplicationServiceWorkerStore,
return entry
entry = yield self.runInteraction(
- "get_oldest_unsent_appservice_txn",
- _get_oldest_unsent_txn,
+ "get_oldest_unsent_appservice_txn", _get_oldest_unsent_txn
)
if not entry:
@@ -312,14 +304,14 @@ class ApplicationServiceTransactionWorkerStore(ApplicationServiceWorkerStore,
events = yield self._get_events(event_ids)
- defer.returnValue(AppServiceTransaction(
- service=service, id=entry["txn_id"], events=events
- ))
+ defer.returnValue(
+ AppServiceTransaction(service=service, id=entry["txn_id"], events=events)
+ )
def _get_last_txn(self, txn, service_id):
txn.execute(
"SELECT last_txn FROM application_services_state WHERE as_id=?",
- (service_id,)
+ (service_id,),
)
last_txn_id = txn.fetchone()
if last_txn_id is None or last_txn_id[0] is None: # no row exists
@@ -332,6 +324,7 @@ class ApplicationServiceTransactionWorkerStore(ApplicationServiceWorkerStore,
txn.execute(
"UPDATE appservice_stream_position SET stream_ordering = ?", (pos,)
)
+
return self.runInteraction(
"set_appservice_last_pos", set_appservice_last_pos_txn
)
@@ -362,7 +355,7 @@ class ApplicationServiceTransactionWorkerStore(ApplicationServiceWorkerStore,
return upper_bound, [row[1] for row in rows]
upper_bound, event_ids = yield self.runInteraction(
- "get_new_events_for_appservice", get_new_events_for_appservice_txn,
+ "get_new_events_for_appservice", get_new_events_for_appservice_txn
)
events = yield self._get_events(event_ids)
|
