Skip to content
This repository has been archived by the owner on Apr 26, 2024. It is now read-only.

Add type hints to purge room and server notice admin API #9520

Merged
merged 6 commits into from
Mar 8, 2021
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changelog.d/9520.misc
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Add type hints to purge room and server notice admin API.
15 changes: 9 additions & 6 deletions synapse/rest/admin/purge_room_servlet.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,13 +12,20 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from typing import TYPE_CHECKING, Tuple

from synapse.http.servlet import (
RestServlet,
assert_params_in_dict,
parse_json_object_from_request,
)
from synapse.http.site import SynapseRequest
from synapse.rest.admin import assert_requester_is_admin
from synapse.rest.admin._base import admin_patterns
from synapse.types import JsonDict

if TYPE_CHECKING:
from synapse.server import HomeServer


class PurgeRoomServlet(RestServlet):
Expand All @@ -36,16 +43,12 @@ class PurgeRoomServlet(RestServlet):

PATTERNS = admin_patterns("/purge_room$")

def __init__(self, hs):
"""
Args:
hs (synapse.server.HomeServer): server
"""
def __init__(self, hs: "HomeServer"):
self.hs = hs
self.auth = hs.get_auth()
self.pagination_handler = hs.get_pagination_handler()

async def on_POST(self, request):
async def on_POST(self, request: SynapseRequest) -> Tuple[int, JsonDict]:
await assert_requester_is_admin(self.auth, request)

body = parse_json_object_from_request(request)
Expand Down
23 changes: 14 additions & 9 deletions synapse/rest/admin/server_notice_servlet.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,17 +12,24 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from typing import TYPE_CHECKING, Optional, Tuple

from synapse.api.constants import EventTypes
from synapse.api.errors import SynapseError
from synapse.http.server import HttpServer
from synapse.http.servlet import (
RestServlet,
assert_params_in_dict,
parse_json_object_from_request,
)
from synapse.http.site import SynapseRequest
from synapse.rest.admin import assert_requester_is_admin
from synapse.rest.admin._base import admin_patterns
from synapse.rest.client.transactions import HttpTransactionCache
from synapse.types import UserID
from synapse.types import JsonDict, UserID

if TYPE_CHECKING:
from synapse.server import HomeServer


class SendServerNoticeServlet(RestServlet):
Expand All @@ -44,17 +51,13 @@ class SendServerNoticeServlet(RestServlet):
}
"""

def __init__(self, hs):
"""
Args:
hs (synapse.server.HomeServer): server
"""
def __init__(self, hs: "HomeServer"):
self.hs = hs
self.auth = hs.get_auth()
self.txns = HttpTransactionCache(hs)
self.snm = hs.get_server_notices_manager()

def register(self, json_resource):
def register(self, json_resource: HttpServer):
PATTERN = "/send_server_notice"
json_resource.register_paths(
"POST", admin_patterns(PATTERN + "$"), self.on_POST, self.__class__.__name__
Expand All @@ -66,7 +69,9 @@ def register(self, json_resource):
self.__class__.__name__,
)

async def on_POST(self, request, txn_id=None):
async def on_POST(
self, request: SynapseRequest, txn_id: Optional[str] = None
) -> Tuple[int, JsonDict]:
await assert_requester_is_admin(self.auth, request)
body = parse_json_object_from_request(request)
assert_params_in_dict(body, ("user_id", "content"))
Expand All @@ -90,7 +95,7 @@ async def on_POST(self, request, txn_id=None):

return 200, {"event_id": event.event_id}

def on_PUT(self, request, txn_id):
def on_PUT(self, request: SynapseRequest, txn_id: str) -> Tuple[int, JsonDict]:
return self.txns.fetch_or_execute_request(
request, self.on_POST, request, txn_id
)