Improve typing syntax and type NamedTuples (#1610)
* Change to PEP 526 typing syntax * Change from collections.namedtuple to typing.NamedTuple
This commit is contained in:
parent
9b928b4679
commit
0c21da8c18
@ -30,7 +30,7 @@ Iterator :class:`NodeIterator` and a context manager
|
|||||||
|
|
||||||
It can be serialized and deserialized with :func:`save_structure_to_file`
|
It can be serialized and deserialized with :func:`save_structure_to_file`
|
||||||
and :func:`load_structure_from_file`, as well as with :mod:`json` and
|
and :func:`load_structure_from_file`, as well as with :mod:`json` and
|
||||||
:mod:`pickle` thanks to being a :func:`~collections.namedtuple`.
|
:mod:`pickle` thanks to being a :class:`~typing.NamedTuple`.
|
||||||
|
|
||||||
``resumable_iteration``
|
``resumable_iteration``
|
||||||
"""""""""""""""""""""""
|
"""""""""""""""""""""""
|
||||||
|
@ -525,7 +525,7 @@ class Instaloader:
|
|||||||
.. versionadded:: 4.3"""
|
.. versionadded:: 4.3"""
|
||||||
|
|
||||||
http_response = self.context.get_raw(url)
|
http_response = self.context.get_raw(url)
|
||||||
date_object = None # type: Optional[datetime]
|
date_object: Optional[datetime] = None
|
||||||
if 'Last-Modified' in http_response.headers:
|
if 'Last-Modified' in http_response.headers:
|
||||||
date_object = datetime.strptime(http_response.headers["Last-Modified"], '%a, %d %b %Y %H:%M:%S GMT')
|
date_object = datetime.strptime(http_response.headers["Last-Modified"], '%a, %d %b %Y %H:%M:%S GMT')
|
||||||
date_object = date_object.replace(tzinfo=timezone.utc)
|
date_object = date_object.replace(tzinfo=timezone.utc)
|
||||||
@ -730,7 +730,7 @@ class Instaloader:
|
|||||||
post.get_sidecar_nodes(self.slide_start, self.slide_end),
|
post.get_sidecar_nodes(self.slide_start, self.slide_end),
|
||||||
start=self.slide_start % post.mediacount + 1
|
start=self.slide_start % post.mediacount + 1
|
||||||
):
|
):
|
||||||
suffix = str(edge_number) # type: Optional[str]
|
suffix: Optional[str] = str(edge_number)
|
||||||
if '{filename}' in self.filename_pattern:
|
if '{filename}' in self.filename_pattern:
|
||||||
suffix = None
|
suffix = None
|
||||||
if self.download_pictures and (not sidecar_node.is_video or self.download_video_thumbnails):
|
if self.download_pictures and (not sidecar_node.is_video or self.download_video_thumbnails):
|
||||||
@ -958,11 +958,11 @@ class Instaloader:
|
|||||||
"""
|
"""
|
||||||
for user_highlight in self.get_highlights(user):
|
for user_highlight in self.get_highlights(user):
|
||||||
name = user_highlight.owner_username
|
name = user_highlight.owner_username
|
||||||
highlight_target = (filename_target
|
highlight_target: Union[str, Path] = (filename_target
|
||||||
if filename_target
|
if filename_target
|
||||||
else (Path(_PostPathFormatter.sanitize_path(name, self.sanitize_paths)) /
|
else (Path(_PostPathFormatter.sanitize_path(name, self.sanitize_paths)) /
|
||||||
_PostPathFormatter.sanitize_path(user_highlight.title,
|
_PostPathFormatter.sanitize_path(user_highlight.title,
|
||||||
self.sanitize_paths))) # type: Union[str, Path]
|
self.sanitize_paths)))
|
||||||
self.context.log("Retrieving highlights \"{}\" from profile {}".format(user_highlight.title, name))
|
self.context.log("Retrieving highlights \"{}\" from profile {}".format(user_highlight.title, name))
|
||||||
self.download_highlight_cover(user_highlight, highlight_target)
|
self.download_highlight_cover(user_highlight, highlight_target)
|
||||||
totalcount = user_highlight.itemcount
|
totalcount = user_highlight.itemcount
|
||||||
|
@ -70,7 +70,7 @@ class InstaloaderContext:
|
|||||||
self.iphone_support = iphone_support
|
self.iphone_support = iphone_support
|
||||||
|
|
||||||
# error log, filled with error() and printed at the end of Instaloader.main()
|
# error log, filled with error() and printed at the end of Instaloader.main()
|
||||||
self.error_log = [] # type: List[str]
|
self.error_log: List[str] = []
|
||||||
|
|
||||||
self._rate_controller = rate_controller(self) if rate_controller is not None else RateController(self)
|
self._rate_controller = rate_controller(self) if rate_controller is not None else RateController(self)
|
||||||
|
|
||||||
@ -81,7 +81,7 @@ class InstaloaderContext:
|
|||||||
self.fatal_status_codes = fatal_status_codes or []
|
self.fatal_status_codes = fatal_status_codes or []
|
||||||
|
|
||||||
# Cache profile from id (mapping from id to Profile)
|
# Cache profile from id (mapping from id to Profile)
|
||||||
self.profile_id_cache = dict() # type: Dict[int, Any]
|
self.profile_id_cache: Dict[int, Any] = dict()
|
||||||
|
|
||||||
@contextmanager
|
@contextmanager
|
||||||
def anonymous_copy(self):
|
def anonymous_copy(self):
|
||||||
@ -589,7 +589,7 @@ class RateController:
|
|||||||
|
|
||||||
def __init__(self, context: InstaloaderContext):
|
def __init__(self, context: InstaloaderContext):
|
||||||
self._context = context
|
self._context = context
|
||||||
self._query_timestamps = dict() # type: Dict[str, List[float]]
|
self._query_timestamps: Dict[str, List[float]] = dict()
|
||||||
self._earliest_next_request_time = 0.0
|
self._earliest_next_request_time = 0.0
|
||||||
self._iphone_earliest_next_request_time = 0.0
|
self._iphone_earliest_next_request_time = 0.0
|
||||||
|
|
||||||
|
@ -10,15 +10,15 @@ from typing import Any, Callable, Dict, Iterable, Iterator, NamedTuple, Optional
|
|||||||
from .exceptions import AbortDownloadException, InvalidArgumentException, QueryReturnedBadRequestException
|
from .exceptions import AbortDownloadException, InvalidArgumentException, QueryReturnedBadRequestException
|
||||||
from .instaloadercontext import InstaloaderContext
|
from .instaloadercontext import InstaloaderContext
|
||||||
|
|
||||||
FrozenNodeIterator = NamedTuple('FrozenNodeIterator',
|
class FrozenNodeIterator(NamedTuple):
|
||||||
[('query_hash', str),
|
query_hash: str
|
||||||
('query_variables', Dict),
|
query_variables: Dict
|
||||||
('query_referer', Optional[str]),
|
query_referer: Optional[str]
|
||||||
('context_username', Optional[str]),
|
context_username: Optional[str]
|
||||||
('total_index', int),
|
total_index: int
|
||||||
('best_before', Optional[float]),
|
best_before: Optional[float]
|
||||||
('remaining_data', Optional[Dict]),
|
remaining_data: Optional[Dict]
|
||||||
('first_node', Optional[Dict])])
|
first_node: Optional[Dict]
|
||||||
FrozenNodeIterator.query_hash.__doc__ = """The GraphQL ``query_hash`` parameter."""
|
FrozenNodeIterator.query_hash.__doc__ = """The GraphQL ``query_hash`` parameter."""
|
||||||
FrozenNodeIterator.query_variables.__doc__ = """The GraphQL ``query_variables`` parameter."""
|
FrozenNodeIterator.query_variables.__doc__ = """The GraphQL ``query_variables`` parameter."""
|
||||||
FrozenNodeIterator.query_referer.__doc__ = """The HTTP referer used for the GraphQL query."""
|
FrozenNodeIterator.query_referer.__doc__ = """The HTTP referer used for the GraphQL query."""
|
||||||
@ -93,7 +93,7 @@ class NodeIterator(Iterator[T]):
|
|||||||
self._first_node: Optional[Dict] = None
|
self._first_node: Optional[Dict] = None
|
||||||
|
|
||||||
def _query(self, after: Optional[str] = None) -> Dict:
|
def _query(self, after: Optional[str] = None) -> Dict:
|
||||||
pagination_variables = {'first': NodeIterator._graphql_page_length} # type: Dict[str, Any]
|
pagination_variables: Dict[str, Any] = {'first': NodeIterator._graphql_page_length}
|
||||||
if after is not None:
|
if after is not None:
|
||||||
pagination_variables['after'] = after
|
pagination_variables['after'] = after
|
||||||
try:
|
try:
|
||||||
|
@ -2,12 +2,11 @@ import json
|
|||||||
import lzma
|
import lzma
|
||||||
import re
|
import re
|
||||||
from base64 import b64decode, b64encode
|
from base64 import b64decode, b64encode
|
||||||
from collections import namedtuple
|
|
||||||
from contextlib import suppress
|
from contextlib import suppress
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from itertools import islice
|
from itertools import islice
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Any, Dict, Iterable, Iterator, List, Optional, Tuple, Union
|
from typing import Any, Dict, Iterable, Iterator, List, NamedTuple, Optional, Tuple, Union
|
||||||
from unicodedata import normalize
|
from unicodedata import normalize
|
||||||
|
|
||||||
from . import __version__
|
from . import __version__
|
||||||
@ -16,25 +15,57 @@ from .instaloadercontext import InstaloaderContext
|
|||||||
from .nodeiterator import FrozenNodeIterator, NodeIterator
|
from .nodeiterator import FrozenNodeIterator, NodeIterator
|
||||||
from .sectioniterator import SectionIterator
|
from .sectioniterator import SectionIterator
|
||||||
|
|
||||||
PostSidecarNode = namedtuple('PostSidecarNode', ['is_video', 'display_url', 'video_url'])
|
|
||||||
PostSidecarNode.__doc__ = "Item of a Sidecar Post."
|
class PostSidecarNode(NamedTuple):
|
||||||
|
"""Item of a Sidecar Post."""
|
||||||
|
is_video: bool
|
||||||
|
display_url: str
|
||||||
|
video_url: str
|
||||||
|
|
||||||
|
|
||||||
PostSidecarNode.is_video.__doc__ = "Whether this node is a video."
|
PostSidecarNode.is_video.__doc__ = "Whether this node is a video."
|
||||||
PostSidecarNode.display_url.__doc__ = "URL of image or video thumbnail."
|
PostSidecarNode.display_url.__doc__ = "URL of image or video thumbnail."
|
||||||
PostSidecarNode.video_url.__doc__ = "URL of video or None."
|
PostSidecarNode.video_url.__doc__ = "URL of video or None."
|
||||||
|
|
||||||
PostCommentAnswer = namedtuple('PostCommentAnswer', ['id', 'created_at_utc', 'text', 'owner', 'likes_count'])
|
|
||||||
|
class PostCommentAnswer(NamedTuple):
|
||||||
|
id: int
|
||||||
|
created_at_utc: datetime
|
||||||
|
text: str
|
||||||
|
owner: 'Profile'
|
||||||
|
likes_count: int
|
||||||
|
|
||||||
|
|
||||||
PostCommentAnswer.id.__doc__ = "ID number of comment."
|
PostCommentAnswer.id.__doc__ = "ID number of comment."
|
||||||
PostCommentAnswer.created_at_utc.__doc__ = ":class:`~datetime.datetime` when comment was created (UTC)."
|
PostCommentAnswer.created_at_utc.__doc__ = ":class:`~datetime.datetime` when comment was created (UTC)."
|
||||||
PostCommentAnswer.text.__doc__ = "Comment text."
|
PostCommentAnswer.text.__doc__ = "Comment text."
|
||||||
PostCommentAnswer.owner.__doc__ = "Owner :class:`Profile` of the comment."
|
PostCommentAnswer.owner.__doc__ = "Owner :class:`Profile` of the comment."
|
||||||
PostCommentAnswer.likes_count.__doc__ = "Number of likes on comment."
|
PostCommentAnswer.likes_count.__doc__ = "Number of likes on comment."
|
||||||
|
|
||||||
PostComment = namedtuple('PostComment', (*PostCommentAnswer._fields, 'answers')) # type: ignore
|
|
||||||
|
class PostComment(NamedTuple):
|
||||||
|
id: int
|
||||||
|
created_at_utc: datetime
|
||||||
|
text: str
|
||||||
|
owner: 'Profile'
|
||||||
|
likes_count: int
|
||||||
|
answers: Iterator[PostCommentAnswer]
|
||||||
|
|
||||||
|
|
||||||
for field in PostCommentAnswer._fields:
|
for field in PostCommentAnswer._fields:
|
||||||
getattr(PostComment, field).__doc__ = getattr(PostCommentAnswer, field).__doc__ # pylint: disable=no-member
|
getattr(PostComment, field).__doc__ = getattr(PostCommentAnswer, field).__doc__ # pylint: disable=no-member
|
||||||
PostComment.answers.__doc__ = r"Iterator which yields all :class:`PostCommentAnswer`\ s for the comment." # type: ignore
|
PostComment.answers.__doc__ = r"Iterator which yields all :class:`PostCommentAnswer`\ s for the comment."
|
||||||
|
|
||||||
|
|
||||||
|
class PostLocation(NamedTuple):
|
||||||
|
id: int
|
||||||
|
name: str
|
||||||
|
slug: str
|
||||||
|
has_public_page: Optional[bool]
|
||||||
|
lat: Optional[float]
|
||||||
|
lng: Optional[float]
|
||||||
|
|
||||||
|
|
||||||
PostLocation = namedtuple('PostLocation', ['id', 'name', 'slug', 'has_public_page', 'lat', 'lng'])
|
|
||||||
PostLocation.id.__doc__ = "ID number of location."
|
PostLocation.id.__doc__ = "ID number of location."
|
||||||
PostLocation.name.__doc__ = "Location name."
|
PostLocation.name.__doc__ = "Location name."
|
||||||
PostLocation.slug.__doc__ = "URL friendly variant of location name."
|
PostLocation.slug.__doc__ = "URL friendly variant of location name."
|
||||||
@ -73,8 +104,8 @@ class Post:
|
|||||||
self._context = context
|
self._context = context
|
||||||
self._node = node
|
self._node = node
|
||||||
self._owner_profile = owner_profile
|
self._owner_profile = owner_profile
|
||||||
self._full_metadata_dict = None # type: Optional[Dict[str, Any]]
|
self._full_metadata_dict: Optional[Dict[str, Any]] = None
|
||||||
self._location = None # type: Optional[PostLocation]
|
self._location: Optional[PostLocation] = None
|
||||||
self._iphone_struct_ = None
|
self._iphone_struct_ = None
|
||||||
if 'iphone_struct' in node:
|
if 'iphone_struct' in node:
|
||||||
# if loaded from JSON with load_structure_from_file()
|
# if loaded from JSON with load_structure_from_file()
|
||||||
@ -516,7 +547,7 @@ class Post:
|
|||||||
def get_comments(self) -> Iterable[PostComment]:
|
def get_comments(self) -> Iterable[PostComment]:
|
||||||
r"""Iterate over all comments of the post.
|
r"""Iterate over all comments of the post.
|
||||||
|
|
||||||
Each comment is represented by a PostComment namedtuple with fields text (string), created_at (datetime),
|
Each comment is represented by a PostComment NamedTuple with fields text (string), created_at (datetime),
|
||||||
id (int), owner (:class:`Profile`) and answers (:class:`~typing.Iterator`\ [:class:`PostCommentAnswer`])
|
id (int), owner (:class:`Profile`) and answers (:class:`~typing.Iterator`\ [:class:`PostCommentAnswer`])
|
||||||
if available.
|
if available.
|
||||||
|
|
||||||
@ -626,7 +657,7 @@ class Post:
|
|||||||
@property
|
@property
|
||||||
def location(self) -> Optional[PostLocation]:
|
def location(self) -> Optional[PostLocation]:
|
||||||
"""
|
"""
|
||||||
If the Post has a location, returns PostLocation namedtuple with fields 'id', 'lat' and 'lng' and 'name'.
|
If the Post has a location, returns PostLocation NamedTuple with fields 'id', 'lat' and 'lng' and 'name'.
|
||||||
|
|
||||||
.. versionchanged:: 4.2.9
|
.. versionchanged:: 4.2.9
|
||||||
Require being logged in (as required by Instagram), return None if not logged-in.
|
Require being logged in (as required by Instagram), return None if not logged-in.
|
||||||
@ -675,7 +706,7 @@ class Profile:
|
|||||||
def __init__(self, context: InstaloaderContext, node: Dict[str, Any]):
|
def __init__(self, context: InstaloaderContext, node: Dict[str, Any]):
|
||||||
assert 'username' in node
|
assert 'username' in node
|
||||||
self._context = context
|
self._context = context
|
||||||
self._has_public_story = None # type: Optional[bool]
|
self._has_public_story: Optional[bool] = None
|
||||||
self._node = node
|
self._node = node
|
||||||
self._has_full_metadata = False
|
self._has_full_metadata = False
|
||||||
self._iphone_struct_ = None
|
self._iphone_struct_ = None
|
||||||
@ -1286,8 +1317,8 @@ class Story:
|
|||||||
def __init__(self, context: InstaloaderContext, node: Dict[str, Any]):
|
def __init__(self, context: InstaloaderContext, node: Dict[str, Any]):
|
||||||
self._context = context
|
self._context = context
|
||||||
self._node = node
|
self._node = node
|
||||||
self._unique_id = None # type: Optional[str]
|
self._unique_id: Optional[str] = None
|
||||||
self._owner_profile = None # type: Optional[Profile]
|
self._owner_profile: Optional[Profile] = None
|
||||||
|
|
||||||
def __repr__(self):
|
def __repr__(self):
|
||||||
return '<Story by {} changed {:%Y-%m-%d_%H-%M-%S_UTC}>'.format(self.owner_username, self.latest_media_utc)
|
return '<Story by {} changed {:%Y-%m-%d_%H-%M-%S_UTC}>'.format(self.owner_username, self.latest_media_utc)
|
||||||
@ -1389,7 +1420,7 @@ class Highlight(Story):
|
|||||||
def __init__(self, context: InstaloaderContext, node: Dict[str, Any], owner: Optional[Profile] = None):
|
def __init__(self, context: InstaloaderContext, node: Dict[str, Any], owner: Optional[Profile] = None):
|
||||||
super().__init__(context, node)
|
super().__init__(context, node)
|
||||||
self._owner_profile = owner
|
self._owner_profile = owner
|
||||||
self._items = None # type: Optional[List[Dict[str, Any]]]
|
self._items: Optional[List[Dict[str, Any]]] = None
|
||||||
|
|
||||||
def __repr__(self):
|
def __repr__(self):
|
||||||
return '<Highlight by {}: {}>'.format(self.owner_username, self.title)
|
return '<Highlight by {}: {}>'.format(self.owner_username, self.title)
|
||||||
|
@ -23,7 +23,7 @@ PRIVATE_PROFILE_ID = 1706625676
|
|||||||
EMPTY_PROFILE = "not_public"
|
EMPTY_PROFILE = "not_public"
|
||||||
EMPTY_PROFILE_ID = 1928659031
|
EMPTY_PROFILE_ID = 1928659031
|
||||||
|
|
||||||
ratecontroller = None # type: Optional[instaloader.RateController]
|
ratecontroller: Optional[instaloader.RateController] = None
|
||||||
|
|
||||||
|
|
||||||
class TestInstaloaderAnonymously(unittest.TestCase):
|
class TestInstaloaderAnonymously(unittest.TestCase):
|
||||||
|
Loading…
Reference in New Issue
Block a user