mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-24 16:30:27 +08:00
Merge pull request #15069 from jmchilton/mypy_unchecked_1
Address some of the new mypy notes about unchecked annotations.
This commit is contained in:
@@ -32,6 +32,7 @@ from typing import (
|
||||
)
|
||||
|
||||
import requests
|
||||
from typing_extensions import TypedDict
|
||||
|
||||
from galaxy import util
|
||||
from galaxy.exceptions import MessageException
|
||||
@@ -102,9 +103,30 @@ class ToolDataPathFiles:
|
||||
return os.path.exists(path)
|
||||
|
||||
|
||||
ErrorListT = List[str]
|
||||
FileNameInfoT = TypedDict(
|
||||
"FileNameInfoT",
|
||||
{
|
||||
"found": bool,
|
||||
"filename": str,
|
||||
"from_shed_config": bool,
|
||||
"tool_data_path": Optional[StrPath],
|
||||
"config_element": Optional[Element],
|
||||
"tool_shed_repository": Optional[Dict[str, Any]],
|
||||
"errors": ErrorListT,
|
||||
},
|
||||
)
|
||||
LoadInfoT = Tuple[Tuple[Element, Optional[StrPath]], Dict[str, Any]]
|
||||
|
||||
|
||||
class ToolDataTable(Dictifiable):
|
||||
type_key: str
|
||||
data: List[List[str]]
|
||||
empty_field_value: str
|
||||
empty_field_values: Dict[Optional[str], str]
|
||||
filenames: Dict[str, FileNameInfoT]
|
||||
_load_info: LoadInfoT
|
||||
_merged_load_info: List[Tuple[Type["ToolDataTable"], LoadInfoT]]
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, d):
|
||||
@@ -131,7 +153,7 @@ class ToolDataTable(Dictifiable):
|
||||
self.empty_field_values: Dict[str, str] = {}
|
||||
self.allow_duplicate_entries = util.asbool(config_element.get("allow_duplicate_entries", True))
|
||||
self.here = os.path.dirname(filename) if filename else None
|
||||
self.filenames: Dict[str, Dict[str, Any]] = {}
|
||||
self.filenames: Dict[str, FileNameInfoT] = {}
|
||||
self.tool_data_path = tool_data_path
|
||||
self.tool_data_path_files = tool_data_path_files
|
||||
self.other_config_dict = other_config_dict or {}
|
||||
@@ -157,7 +179,7 @@ class ToolDataTable(Dictifiable):
|
||||
self._loaded_content_version += 1
|
||||
return self._loaded_content_version
|
||||
|
||||
def get_empty_field_by_name(self, name):
|
||||
def get_empty_field_by_name(self, name: Optional[str]) -> str:
|
||||
return self.empty_field_values.get(name, self.empty_field_value)
|
||||
|
||||
def _add_entry(
|
||||
@@ -355,7 +377,7 @@ class TabularToolDataTable(ToolDataTable):
|
||||
filename = f"{corrected_filename}.sample"
|
||||
found = True
|
||||
|
||||
errors: List[str] = []
|
||||
errors: ErrorListT = []
|
||||
if found:
|
||||
self.extend_data_with(filename, errors=errors)
|
||||
self._update_version()
|
||||
@@ -478,14 +500,14 @@ class TabularToolDataTable(ToolDataTable):
|
||||
if "name" not in self.columns:
|
||||
self.columns["name"] = self.columns["value"]
|
||||
|
||||
def extend_data_with(self, filename: str, errors: Optional[List[str]] = None) -> None:
|
||||
def extend_data_with(self, filename: str, errors: Optional[ErrorListT] = None) -> None:
|
||||
here = os.path.dirname(os.path.abspath(filename))
|
||||
self.data.extend(self.parse_file_fields(filename, errors=errors, here=here))
|
||||
if not self.allow_duplicate_entries:
|
||||
self._deduplicate_data()
|
||||
|
||||
def parse_file_fields(
|
||||
self, filename: str, errors: Optional[List[str]] = None, here: str = "__HERE__"
|
||||
self, filename: str, errors: Optional[ErrorListT] = None, here: str = "__HERE__"
|
||||
) -> List[List[str]]:
|
||||
"""
|
||||
Parse separated lines from file and return a list of tuples.
|
||||
|
||||
@@ -16,6 +16,7 @@ from galaxy.tool_util.parser.util import (
|
||||
DEFAULT_DELTA_FRAC,
|
||||
)
|
||||
from galaxy.util import (
|
||||
Element,
|
||||
ElementTree,
|
||||
string_as_bool,
|
||||
xml_text,
|
||||
@@ -56,6 +57,7 @@ class XmlToolSource(ToolSource):
|
||||
"""Responsible for parsing a tool from classic Galaxy representation."""
|
||||
|
||||
language = "xml"
|
||||
root: Element
|
||||
|
||||
def __init__(self, xml_tree: ElementTree, source_path=None, macro_paths=None):
|
||||
self.xml_tree = xml_tree
|
||||
|
||||
@@ -1,9 +1,11 @@
|
||||
import logging
|
||||
from typing import Set
|
||||
|
||||
from sqlalchemy import (
|
||||
false,
|
||||
func,
|
||||
)
|
||||
from typing_extensions import TypedDict
|
||||
|
||||
from galaxy import (
|
||||
model,
|
||||
@@ -17,6 +19,7 @@ from galaxy.exceptions import (
|
||||
from galaxy.managers.quotas import QuotaManager
|
||||
from galaxy.model import tool_shed_install as install_model
|
||||
from galaxy.security.validate_user_input import validate_password
|
||||
from galaxy.structured_app import StructuredApp
|
||||
from galaxy.util import (
|
||||
nice_size,
|
||||
pretty_print_time_interval,
|
||||
@@ -535,6 +538,10 @@ class ToolVersionListGrid(grids.Grid):
|
||||
return trans.install_model.context.query(self.model_class)
|
||||
|
||||
|
||||
# TODO: Convert admin UI to use the API and drop this.
|
||||
DatatypesEntryT = TypedDict("DatatypesEntryT", {"status": str, "keys": list, "data": list, "message": str})
|
||||
|
||||
|
||||
class AdminGalaxy(controller.JSAppLauncher):
|
||||
|
||||
user_list_grid = UserListGrid()
|
||||
@@ -564,7 +571,7 @@ class AdminGalaxy(controller.JSAppLauncher):
|
||||
"Resend Activation Email", condition=(lambda item: not item.active), allow_multiple=False
|
||||
)
|
||||
|
||||
def __init__(self, app):
|
||||
def __init__(self, app: StructuredApp):
|
||||
super().__init__(app)
|
||||
self.quota_manager: QuotaManager = QuotaManager(app)
|
||||
|
||||
@@ -594,9 +601,9 @@ class AdminGalaxy(controller.JSAppLauncher):
|
||||
@web.expose
|
||||
@web.json
|
||||
@web.require_admin
|
||||
def data_types_list(self, trans, **kwd):
|
||||
def data_types_list(self, trans, **kwd) -> DatatypesEntryT:
|
||||
datatypes = []
|
||||
keys = set()
|
||||
keys: Set[str] = set()
|
||||
message = kwd.get("message", "")
|
||||
status = kwd.get("status", "done")
|
||||
for dtype in sorted(trans.app.datatypes_registry.datatype_elems, key=lambda dt: dt.get("extension")):
|
||||
|
||||
@@ -202,7 +202,7 @@ class TestFolderContentsApi(ApiTestCase):
|
||||
self._assert_index_count_is_correct(response, expected_contents_count=1)
|
||||
|
||||
@requires_new_library
|
||||
def test_index_permissions_include_deleted(self, history_id):
|
||||
def test_index_permissions_include_deleted(self, history_id) -> None:
|
||||
folder_name = "Test Folder Contents Index permissions include deleted"
|
||||
folder_id = self._create_folder_in_library(folder_name)
|
||||
|
||||
|
||||
@@ -239,7 +239,7 @@ class TestMappings(BaseModelTestCase):
|
||||
for i in range(elements):
|
||||
assert c1[i] == dces[i]
|
||||
|
||||
def test_dataset_instance_order(self):
|
||||
def test_dataset_instance_order(self) -> None:
|
||||
u = model.User(email="mary@example.com", password="password")
|
||||
h1 = model.History(name="History 1", user=u)
|
||||
elements = []
|
||||
|
||||
@@ -257,7 +257,7 @@ class MapOverTestCase(NamedTuple):
|
||||
steps: Dict[int, Any]
|
||||
|
||||
|
||||
def _construct_steps_for_map_over():
|
||||
def _construct_steps_for_map_over() -> List[MapOverTestCase]:
|
||||
test_case = MapOverTestCase
|
||||
# these are the cartesian product of
|
||||
# data_input = ['dataset', 'list', 'list:pair', 'list:list']
|
||||
|
||||
Reference in New Issue
Block a user