Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Add initial type hints #128

Merged
merged 8 commits into from
Oct 14, 2022
Merged

Add initial type hints #128

Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
6b627b8
Minor generic cleanup
PrOF-kk Feb 26, 2022
910b7c6
Remove redundant u string prefix
PrOF-kk Feb 27, 2022
d158fa5
Replace some str.format with fstrings
PrOF-kk Feb 27, 2022
7f9d0f4
Fix Keep::dump docs
PrOF-kk Aug 27, 2022
ebd0f5c
Add initial type hints
PrOF-kk Aug 27, 2022
49e72a2
Remove Python 2 regular expression support, add typing
PrOF-kk Sep 27, 2022
0ab21a4
Remove unnecessary typing info from docstrings
PrOF-kk Sep 27, 2022
00f2017
Merge branch 'deprecate_py2' into initialTyping
kiwiz Oct 2, 2022
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
159 changes: 81 additions & 78 deletions gkeepapi/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
import re
import time
import random
from typing import Callable, Iterator, List, Optional, Tuple, Dict, Union

from uuid import getnode as get_mac

Expand All @@ -20,11 +21,6 @@

logger = logging.getLogger(__name__)

try:
Pattern = re._pattern_type # pylint: disable=protected-access
except AttributeError:
Pattern = re.Pattern # pylint: disable=no-member

class APIAuth(object):
"""Authentication token manager"""
def __init__(self, scopes):
Expand Down Expand Up @@ -653,7 +649,7 @@ class Keep(object):
"""
OAUTH_SCOPES = 'oauth2:https://www.googleapis.com/auth/memento https://www.googleapis.com/auth/reminders'

def __init__(self):
def __init__(self) -> None:
self._keep_api = KeepAPI()
self._reminders_api = RemindersAPI()
self._media_api = MediaAPI()
Expand All @@ -675,13 +671,13 @@ def _clear(self):
root_node = _node.Root()
self._nodes[_node.Root.ID] = root_node

def login(self, email, password, state=None, sync=True, device_id=None):
def login(self, email: str, password: str, state: Optional[Dict] = None, sync=True, device_id: Optional[str] = None):
"""Authenticate to Google with the provided credentials & sync.

Args:
email (str): The account to use.
password (str): The account password.
state (dict): Serialized state to load.
email: The account to use.
password: The account password.
state: Serialized state to load.

Raises:
LoginException: If there was a problem logging in.
Expand All @@ -690,19 +686,18 @@ def login(self, email, password, state=None, sync=True, device_id=None):
if device_id is None:
device_id = get_mac()

ret = auth.login(email, password, device_id)
if ret:
self.load(auth, state, sync)
auth.login(email, password, device_id)
self.load(auth, state, sync)

return ret
return True

def resume(self, email, master_token, state=None, sync=True, device_id=None):
def resume(self, email: str, master_token: str, state: Optional[Dict] = None, sync=True, device_id: Optional[int] = None):
"""Authenticate to Google with the provided master token & sync.

Args:
email (str): The account to use.
master_token (str): The master token.
state (dict): Serialized state to load.
email: The account to use.
master_token: The master token.
state: Serialized state to load.

Raises:
LoginException: If there was a problem logging in.
Expand All @@ -711,25 +706,24 @@ def resume(self, email, master_token, state=None, sync=True, device_id=None):
if device_id is None:
device_id = get_mac()

ret = auth.load(email, master_token, device_id)
if ret:
self.load(auth, state, sync)
auth.load(email, master_token, device_id)
self.load(auth, state, sync)

return ret
return True

def getMasterToken(self):
def getMasterToken(self) -> str:
"""Get master token for resuming.

Returns:
str: The master token.
The master token.
"""
return self._keep_api.getAuth().getMasterToken()

def load(self, auth, state=None, sync=True):
def load(self, auth: APIAuth, state: Optional[Dict] = None, sync=True) -> None:
"""Authenticate to Google with a prepared authentication object & sync.
Args:
auth (APIAuth): Authentication object.
state (dict): Serialized state to load.
auth: Authentication object.
state: Serialized state to load.

Raises:
LoginException: If there was a problem logging in.
Expand All @@ -742,11 +736,11 @@ def load(self, auth, state=None, sync=True):
if sync:
self.sync(True)

def dump(self):
def dump(self) -> Dict:
"""Serialize note data.

Args:
state (dict): Serialized state to load.
Returns:
Serialized state.
"""
# Find all nodes manually, as the Keep object isn't aware of new
# ListItems until they've been synced to the server.
Expand All @@ -761,61 +755,70 @@ def dump(self):
'nodes': [node.save(False) for node in nodes]
}

def restore(self, state):
def restore(self, state: Dict) -> None:
"""Unserialize saved note data.

Args:
state (dict): Serialized state to load.
state: Serialized state to load.
"""
self._clear()
self._parseUserInfo({'labels': state['labels']})
self._parseNodes(state['nodes'])
self._keep_version = state['keep_version']

def get(self, node_id):
def get(self, node_id: str) -> _node.TopLevelNode:
"""Get a note with the given ID.

Args:
node_id (str): The note ID.
node_id: The note ID.

Returns:
gkeepapi.node.TopLevelNode: The Note or None if not found.
The Note or None if not found.
"""
return \
self._nodes[_node.Root.ID].get(node_id) or \
self._nodes[_node.Root.ID].get(self._sid_map.get(node_id))

def add(self, node):
def add(self, node: _node.Node) -> None:
"""Register a top level node (and its children) for syncing up to the server. There's no need to call this for nodes created by
:py:meth:`createNote` or :py:meth:`createList` as they are automatically added.

LoginException: If :py:meth:`login` has not been called.
Args:
node (gkeepapi.node.Node): The node to sync.
node: The node to sync.

Raises:
Invalid: If the parent node is not found.
InvalidException: If the parent node is not found.
"""
if node.parent_id != _node.Root.ID:
raise exception.InvalidException('Not a top level node')

self._nodes[node.id] = node
self._nodes[node.parent_id].append(node, False)

def find(self, query=None, func=None, labels=None, colors=None, pinned=None, archived=None, trashed=False): # pylint: disable=too-many-arguments
def find(
self,
query: Union[re.Pattern, str, None] = None,
func: Optional[Callable] = None,
labels: Optional[List[str]] = None,
colors: Optional[List[str]] = None,
pinned: Optional[bool] = None,
archived: Optional[bool] = None,
trashed: Optional[bool] = False
) -> Iterator[_node.TopLevelNode]: # pylint: disable=too-many-arguments
"""Find Notes based on the specified criteria.

Args:
query (Union[_sre.SRE_Pattern, str, None]): A str or regular expression to match against the title and text.
func (Union[callable, None]): A filter function.
labels (Union[List[str], None]): A list of label ids or objects to match. An empty list matches notes with no labels.
colors (Union[List[str], None]): A list of colors to match.
pinned (Union[bool, None]): Whether to match pinned notes.
archived (Union[bool, None]): Whether to match archived notes.
trashed (Union[bool, None]): Whether to match trashed notes.
query: A str or regular expression to match against the title and text.
func: A filter function.
labels: A list of label ids or objects to match. An empty list matches notes with no labels.
colors: A list of colors to match.
pinned: Whether to match pinned notes.
archived: Whether to match archived notes.
trashed: Whether to match trashed notes.

Return:
Generator[gkeepapi.node.TopLevelNode]: Results.
Search results.
"""
if labels is not None:
labels = [i.id if isinstance(i, _node.Label) else i for i in labels]
Expand All @@ -824,7 +827,7 @@ def find(self, query=None, func=None, labels=None, colors=None, pinned=None, arc
# Process the query.
(query is None or (
(isinstance(query, str) and (query in node.title or query in node.text)) or
(isinstance(query, Pattern) and (
(isinstance(query, re.Pattern) and (
query.search(node.title) or query.search(node.text)
))
)) and
Expand All @@ -845,15 +848,15 @@ def find(self, query=None, func=None, labels=None, colors=None, pinned=None, arc
(trashed is None or node.trashed == trashed)
)

def createNote(self, title=None, text=None):
def createNote(self, title: Optional[str] = None, text: Optional[str] = None) -> _node.Node:
"""Create a new managed note. Any changes to the note will be uploaded when :py:meth:`sync` is called.

Args:
title (str): The title of the note.
text (str): The text of the note.
title: The title of the note.
text: The text of the note.

Returns:
gkeepapi.node.List: The new note.
The new note.
"""
node = _node.Note()
if title is not None:
Expand All @@ -863,15 +866,15 @@ def createNote(self, title=None, text=None):
self.add(node)
return node

def createList(self, title=None, items=None):
def createList(self, title: Optional[str] = None, items: Optional[List[Tuple[str, bool]]] = None) -> _node.List:
"""Create a new list and populate it. Any changes to the note will be uploaded when :py:meth:`sync` is called.

Args:
title (str): The title of the list.
items (List[(str, bool)]): A list of tuples. Each tuple represents the text and checked status of the listitem.
title: The title of the list.
items: A list of tuples. Each tuple represents the text and checked status of the listitem.

Returns:
gkeepapi.node.List: The new list.
The new list.
"""
if items is None:
items = []
Expand All @@ -887,14 +890,14 @@ def createList(self, title=None, items=None):
self.add(node)
return node

def createLabel(self, name):
def createLabel(self, name: str) -> _node.Label:
"""Create a new label.

Args:
name (str): Label name.
name: Label name.

Returns:
gkeepapi.node.Label: The new label.
The new label.

Raises:
LabelException: If the label exists.
Expand All @@ -906,15 +909,15 @@ def createLabel(self, name):
self._labels[node.id] = node # pylint: disable=protected-access
return node

def findLabel(self, query, create=False):
def findLabel(self, query: Union[re.Pattern, str], create=False) -> Optional[_node.Label]:
"""Find a label with the given name.

Args:
name (Union[_sre.SRE_Pattern, str]): A str or regular expression to match against the name.
create (bool): Whether to create the label if it doesn't exist (only if name is a str).
name: A str or regular expression to match against the name.
create: Whether to create the label if it doesn't exist (only if name is a str).

Returns:
Union[gkeepapi.node.Label, None]: The label.
The label.
"""
is_str = isinstance(query, str)
name = None
Expand All @@ -925,27 +928,27 @@ def findLabel(self, query, create=False):
for label in self._labels.values():
# Match the label against query, which may be a str or Pattern.
if (is_str and query == label.name.lower()) or \
(isinstance(query, Pattern) and query.search(label.name)):
(isinstance(query, re.Pattern) and query.search(label.name)):
return label

return self.createLabel(name) if create and is_str else None

def getLabel(self, label_id):
def getLabel(self, label_id: str) -> Optional[_node.Label]:
"""Get an existing label.

Args:
label_id (str): Label id.
label_id: Label id.

Returns:
Union[gkeepapi.node.Label, None]: The label.
The label.
"""
return self._labels.get(label_id)

def deleteLabel(self, label_id):
def deleteLabel(self, label_id: str) -> None:
"""Deletes a label.

Args:
label_id (str): Label id.
label_id: Label id.
"""
if label_id not in self._labels:
return
Expand All @@ -955,38 +958,38 @@ def deleteLabel(self, label_id):
for node in self.all():
node.labels.remove(label)

def labels(self):
def labels(self) -> List[_node.Label]:
"""Get all labels.

Returns:
List[gkeepapi.node.Label]: Labels
Labels
"""
return self._labels.values()

def getMediaLink(self, blob):
def getMediaLink(self, blob: _node.Blob) -> str:
"""Get the canonical link to media.

Args:
blob (gkeepapi.node.Blob): The media resource.
blob: The media resource.

Returns:
str: A link to the media.
A link to the media.
"""
return self._media_api.get(blob)

def all(self):
def all(self) -> List[_node.TopLevelNode]:
"""Get all Notes.

Returns:
List[gkeepapi.node.TopLevelNode]: Notes
Notes
"""
return self._nodes[_node.Root.ID].children

def sync(self, resync=False):
def sync(self, resync=False) -> None:
"""Sync the local Keep tree with the server. If resyncing, local changes will be destroyed. Otherwise, local changes to notes, labels and reminders will be detected and synced up.

Args:
resync (bool): Whether to resync data.
resync: Whether to resync data.

Raises:
SyncException: If there is a consistency issue.
Expand Down