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

Add type hints to media rest resources. #9093

Merged
merged 8 commits into from
Jan 15, 2021
Merged
50 changes: 34 additions & 16 deletions synapse/rest/media/v1/filepath.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
# -*- coding: utf-8 -*-
# Copyright 2014-2016 OpenMarket Ltd
# Copyright 2020-2021 The Matrix.org Foundation C.I.C.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
Expand All @@ -16,11 +17,12 @@
import functools
import os
import re
from typing import Callable, List

NEW_FORMAT_ID_RE = re.compile(r"^\d\d\d\d-\d\d-\d\d")


def _wrap_in_base_path(func):
def _wrap_in_base_path(func: "Callable[..., str]") -> "Callable[..., str]":
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why the quotes?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I believe that ... breaks on Python 3.5? We seem to quote similar things in synapse/storage/database.py, but not other places...

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

# Python 3.5.2 doesn't support Callable with an ellipsis, so we wrap it in quotes so
# that mypy sees the type but the runtime python doesn't.

"""Takes a function that returns a relative path and turns it into an
absolute path based on the location of the primary media store
"""
Expand All @@ -41,12 +43,18 @@ class MediaFilePaths:
to write to the backup media store (when one is configured)
"""

def __init__(self, primary_base_path):
def __init__(self, primary_base_path: str):
self.base_path = primary_base_path

def default_thumbnail_rel(
self, default_top_level, default_sub_type, width, height, content_type, method
):
self,
default_top_level: str,
default_sub_type: str,
width: int,
height: int,
content_type: str,
method: str,
) -> str:
top_level_type, sub_type = content_type.split("/")
file_name = "%i-%i-%s-%s-%s" % (width, height, top_level_type, sub_type, method)
return os.path.join(
Expand All @@ -55,12 +63,14 @@ def default_thumbnail_rel(

default_thumbnail = _wrap_in_base_path(default_thumbnail_rel)

def local_media_filepath_rel(self, media_id):
def local_media_filepath_rel(self, media_id: str) -> str:
return os.path.join("local_content", media_id[0:2], media_id[2:4], media_id[4:])

local_media_filepath = _wrap_in_base_path(local_media_filepath_rel)

def local_media_thumbnail_rel(self, media_id, width, height, content_type, method):
def local_media_thumbnail_rel(
self, media_id: str, width: int, height: int, content_type: str, method: str
) -> str:
top_level_type, sub_type = content_type.split("/")
file_name = "%i-%i-%s-%s-%s" % (width, height, top_level_type, sub_type, method)
return os.path.join(
Expand All @@ -86,16 +96,22 @@ def local_media_thumbnail_dir(self, media_id: str) -> str:
media_id[4:],
)

def remote_media_filepath_rel(self, server_name, file_id):
def remote_media_filepath_rel(self, server_name: str, file_id: str) -> str:
return os.path.join(
"remote_content", server_name, file_id[0:2], file_id[2:4], file_id[4:]
)

remote_media_filepath = _wrap_in_base_path(remote_media_filepath_rel)

def remote_media_thumbnail_rel(
self, server_name, file_id, width, height, content_type, method
):
self,
server_name: str,
file_id: str,
width: int,
height: int,
content_type: str,
method: str,
) -> str:
top_level_type, sub_type = content_type.split("/")
file_name = "%i-%i-%s-%s-%s" % (width, height, top_level_type, sub_type, method)
return os.path.join(
Expand All @@ -113,7 +129,7 @@ def remote_media_thumbnail_rel(
# Should be removed after some time, when most of the thumbnails are stored
# using the new path.
def remote_media_thumbnail_rel_legacy(
self, server_name, file_id, width, height, content_type
self, server_name: str, file_id: str, width: int, height: int, content_type: str
):
top_level_type, sub_type = content_type.split("/")
file_name = "%i-%i-%s-%s" % (width, height, top_level_type, sub_type)
Expand All @@ -126,7 +142,7 @@ def remote_media_thumbnail_rel_legacy(
file_name,
)

def remote_media_thumbnail_dir(self, server_name, file_id):
def remote_media_thumbnail_dir(self, server_name: str, file_id: str) -> str:
return os.path.join(
self.base_path,
"remote_thumbnail",
Expand All @@ -136,7 +152,7 @@ def remote_media_thumbnail_dir(self, server_name, file_id):
file_id[4:],
)

def url_cache_filepath_rel(self, media_id):
def url_cache_filepath_rel(self, media_id: str) -> str:
if NEW_FORMAT_ID_RE.match(media_id):
# Media id is of the form <DATE><RANDOM_STRING>
# E.g.: 2017-09-28-fsdRDt24DS234dsf
Expand All @@ -146,7 +162,7 @@ def url_cache_filepath_rel(self, media_id):

url_cache_filepath = _wrap_in_base_path(url_cache_filepath_rel)

def url_cache_filepath_dirs_to_delete(self, media_id):
def url_cache_filepath_dirs_to_delete(self, media_id: str) -> List[str]:
"The dirs to try and remove if we delete the media_id file"
if NEW_FORMAT_ID_RE.match(media_id):
return [os.path.join(self.base_path, "url_cache", media_id[:10])]
Expand All @@ -156,7 +172,9 @@ def url_cache_filepath_dirs_to_delete(self, media_id):
os.path.join(self.base_path, "url_cache", media_id[0:2]),
]

def url_cache_thumbnail_rel(self, media_id, width, height, content_type, method):
def url_cache_thumbnail_rel(
self, media_id: str, width: int, height: int, content_type: str, method: str
) -> str:
# Media id is of the form <DATE><RANDOM_STRING>
# E.g.: 2017-09-28-fsdRDt24DS234dsf

Expand All @@ -178,7 +196,7 @@ def url_cache_thumbnail_rel(self, media_id, width, height, content_type, method)

url_cache_thumbnail = _wrap_in_base_path(url_cache_thumbnail_rel)

def url_cache_thumbnail_directory(self, media_id):
def url_cache_thumbnail_directory(self, media_id: str) -> str:
# Media id is of the form <DATE><RANDOM_STRING>
# E.g.: 2017-09-28-fsdRDt24DS234dsf

Expand All @@ -195,7 +213,7 @@ def url_cache_thumbnail_directory(self, media_id):
media_id[4:],
)

def url_cache_thumbnail_dirs_to_delete(self, media_id):
def url_cache_thumbnail_dirs_to_delete(self, media_id: str) -> List[str]:
"The dirs to try and remove if we delete the media_id thumbnails"
# Media id is of the form <DATE><RANDOM_STRING>
# E.g.: 2017-09-28-fsdRDt24DS234dsf
Expand Down