Fix return type annotation of dataset size functions

Fix the following mypy error (when using `strict_equality = True`):

```
lib/galaxy/datatypes/tabular.py:552: error: Non-overlapping equality check (left operand type: "int", right operand type: "str")  [comparison-overlap]
                            if dataset_fh.tell() != dataset.get_size():
```
This commit is contained in:
Nicola Soranzo
2023-03-15 17:01:02 +00:00
parent 1dc25d1bbd
commit f45d307db1
7 changed files with 22 additions and 14 deletions
+1 -1
View File
@@ -86,7 +86,7 @@ class DatasetProtocol(
def get_mime(self) -> str:
...
def get_size(self) -> str:
def get_size(self) -> int:
...
def has_data(self) -> bool:
+14 -4
View File
@@ -27,6 +27,7 @@ from typing import (
List,
NamedTuple,
Optional,
overload,
Set,
Tuple,
Type,
@@ -102,6 +103,7 @@ from sqlalchemy.orm import (
from sqlalchemy.orm.collections import attribute_mapped_collection
from sqlalchemy.sql import exists
from typing_extensions import (
Literal,
Protocol,
TypedDict,
)
@@ -3915,16 +3917,24 @@ class Dataset(Base, StorableObject, Serializable):
def _extra_files_rel_path(self):
return self._extra_files_path or self.extra_files_path_name
def _calculate_size(self):
def _calculate_size(self) -> int:
if self.external_filename:
try:
return os.path.getsize(self.external_filename)
except OSError:
return 0
else:
return self.object_store.size(self)
assert self.object_store
return self.object_store.size(self)
def get_size(self, nice_size=False, calculate_size=True):
@overload
def get_size(self, nice_size: Literal[False], calculate_size: bool = True) -> int:
...
@overload
def get_size(self, nice_size: Literal[True], calculate_size: bool = True) -> str:
...
def get_size(self, nice_size: bool = False, calculate_size: bool = True) -> Union[int, str]:
"""Returns the size of the data on disk"""
if self.file_size:
if nice_size:
+2 -2
View File
@@ -170,7 +170,7 @@ class ObjectStore(metaclass=abc.ABCMeta):
raise NotImplementedError()
@abc.abstractmethod
def size(self, obj, extra_dir=None, extra_dir_at_root=False, alt_name=None, obj_dir=False):
def size(self, obj, extra_dir=None, extra_dir_at_root=False, alt_name=None, obj_dir=False) -> int:
"""
Return size of the object identified by `obj`.
@@ -850,7 +850,7 @@ class DiskObjectStore(ConcreteObjectStore):
"""Override `ObjectStore`'s stub by checking file size on disk."""
return self.size(obj, **kwargs) == 0
def _size(self, obj, **kwargs):
def _size(self, obj, **kwargs) -> int:
"""Override `ObjectStore`'s stub by return file size on disk.
Returns 0 if the object doesn't exist yet or other error.
+1 -2
View File
@@ -400,8 +400,7 @@ class Cloud(ConcreteObjectStore, CloudConfigMixin):
def _get_size_in_cloud(self, rel_path):
try:
obj = self.bucket.objects.get(rel_path)
if obj:
return obj.size
return obj.size
except Exception:
log.exception("Could not get size of key '%s' from S3", rel_path)
return -1
+1 -1
View File
@@ -608,7 +608,7 @@ class IRODSObjectStore(DiskObjectStore, CloudConfigMixin):
else:
raise ObjectNotFound(f"objectstore.empty, object does not exist: {obj}, kwargs: {kwargs}")
def _size(self, obj, **kwargs):
def _size(self, obj, **kwargs) -> int:
ipt_timer = ExecutionTimer()
rel_path = self._construct_path(obj, **kwargs)
if self._in_cache(rel_path):
+1 -1
View File
@@ -309,7 +309,7 @@ class PithosObjectStore(ConcreteObjectStore):
raise ObjectNotFound(f"objectstore.empty, object does not exist: {obj}, kwargs: {kwargs}")
return bool(self._size(obj, **kwargs))
def _size(self, obj, **kwargs):
def _size(self, obj, **kwargs) -> int:
"""
:returns: The size of the object, or 0 if it doesn't exist (sorry for
that, not our fault, the ObjectStore interface is like that some
+2 -3
View File
@@ -385,9 +385,8 @@ class S3ObjectStore(ConcreteObjectStore, CloudConfigMixin):
def _get_size_in_s3(self, rel_path):
try:
key = self._bucket.get_key(rel_path)
if key:
return key.size
except S3ResponseError:
return key.size
except (S3ResponseError, AttributeError):
log.exception("Could not get size of key '%s' from S3", rel_path)
return -1