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.

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

Fix most of the typing errors in datastore.py #1133

Merged
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
51 changes: 32 additions & 19 deletions GTG/core/datastore.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,10 +35,12 @@
from GTG.backends.backend_signals import BackendSignals
from GTG.backends.generic_backend import GenericBackend
import GTG.core.info as info
import GTG.core.dirs as dirs
import GTG.core.versioning as versioning

from lxml import etree as et

from typing import Optional
from typing import Optional, Dict


log = logging.getLogger(__name__)
Expand All @@ -54,15 +56,15 @@ def __init__(self) -> None:
self.tasks = TaskStore()
self.tags = TagStore()
self.saved_searches = SavedSearchStore()
self.xml_tree = None
self.xml_tree: Optional[et._ElementTree] = None

self._mutex = threading.Lock()
self.backends = {}
self.backends: Dict[str,GenericBackend] = {}
self._backend_signals = BackendSignals()

# When a backup has to be used, this will be filled with
# info on the backup used
self.backup_info = {}
self.backup_info: Dict[str,str] = {}

# Flag when turned to true, all pending operation should be
# completed and then GTG should quit
Expand All @@ -75,7 +77,7 @@ def __init__(self) -> None:
'closed': {'all': 0, 'untagged': 0},
}

self.data_path = None
self.data_path: Optional[str] = None
self._activate_non_default_backends()


Expand All @@ -84,20 +86,28 @@ def mutex(self) -> threading.Lock:
return self._mutex


def load_data(self, data: et.Element) -> None:
def load_data(self, data: et._ElementTree) -> None:
"""Load data from an lxml element object."""

self.saved_searches.from_xml(data.find('searchlist'))
self.tags.from_xml(data.find('taglist'))
self.tasks.from_xml(data.find('tasklist'), self.tags)
searches_xml = data.find('searchlist')
tags_xml = data.find('taglist')
tasks_xml = data.find('tasklist')

assert searches_xml is not None, "Missing 'searchlist' tag in xml file."
assert tags_xml is not None, "Missing 'taglist' tag in xml file."
assert tasks_xml is not None, "Missing 'tasklist' tag in xml file."

self.saved_searches.from_xml(searches_xml)
self.tags.from_xml(tags_xml)
self.tasks.from_xml(tasks_xml, self.tags)

self.refresh_task_count()


def load_file(self, path: str) -> None:
"""Load data from a file."""

bench_start = 0
bench_start = 0.0

if log.isEnabledFor(logging.DEBUG):
bench_start = time()
Expand All @@ -106,6 +116,7 @@ def load_file(self, path: str) -> None:

with open(path, 'rb') as stream:
self.xml_tree = et.parse(stream, parser=parser)
assert isinstance(self.xml_tree, et._ElementTree), 'Parsing should return an _ElementTree object'
self.load_data(self.xml_tree)

if log.isEnabledFor(logging.DEBUG):
Expand All @@ -116,7 +127,7 @@ def load_file(self, path: str) -> None:
self.data_path = path


def generate_xml(self) -> et.ElementTree:
def generate_xml(self) -> et._ElementTree:
"""Generate lxml element object with all data."""

root = et.Element('gtgData')
Expand Down Expand Up @@ -144,8 +155,9 @@ def save(self, path: Optional[str] = None) -> None:
"""Write GTG data file."""

path = path or self.data_path
assert path is not None, "Failed to determine save location."
temp_file = path + '__'
bench_start = 0
bench_start = 0.0

try:
os.rename(path, temp_file)
Expand Down Expand Up @@ -268,7 +280,7 @@ def notify_tag_change(self, tag) -> None:
task.notify('show_tag_colors')


def first_run(self, path: str) -> et.Element:
def first_run(self, path: str) -> None:
"""Write initial data file."""

self.xml_tree = firstrun_tasks.generate()
Expand All @@ -279,16 +291,17 @@ def first_run(self, path: str) -> et.Element:
def do_first_run_versioning(self, filepath: str) -> None:
"""If there is an old file around needing versioning, convert it, then rename the old file."""

old_path = self.find_old_path(DATA_DIR)
old_path = self.find_old_path(dirs.DATA_DIR)

if old_path is not None:
log.warning('Found old file: %r. Running versioning code.', old_path)
tree = versioning.convert(old_path, self)
tree = versioning.convert(old_path)
self.load_data(tree)
self.save(filepath)
os.rename(old_path, old_path + '.imported')

else:
assert self.data_path is not None, "Save location not set."
self.first_run(self.data_path)


Expand Down Expand Up @@ -319,16 +332,16 @@ def find_old_uuid_path(self, path: str) -> Optional[str]:
for backend in xml_tree.findall('backend'):
module = backend.get('module')
if module == 'backend_localfile':
uuid_path = backend.get('path')
uuid_path: Optional[str] = backend.get('path')
assert uuid_path is not None, "Missing 'path' propoerty in backend."
if os.path.isfile(uuid_path):

return uuid_path

return None


@staticmethod
def get_backup_path(path: str, i: int = None) -> str:
def get_backup_path(path: str, i: Optional[int] = None) -> str:
"""Get path of backups which are backup/ directory."""

dirname, filename = os.path.split(path)
Expand Down Expand Up @@ -685,7 +698,7 @@ def _internal_flush_all_tasks():
def fill_with_samples(self, tasks_count: int) -> None:
"""Fill the Datastore with sample data."""

def random_date(start: datetime = None):
def random_date(start: Optional[datetime] = None):
start = start or datetime.now()
end = start + timedelta(days=random.randint(1, 365 * 5))

Expand Down
2 changes: 1 addition & 1 deletion GTG/core/dirs.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
"""
import os

from gi.repository import GLib
from gi.repository import GLib # type: ignore[import-untyped]

# Folder where core GTG data is stored like services information, tags, etc
DATA_DIR = os.path.join(GLib.get_user_data_dir(), 'gtg')
Expand Down