refactor: replace hidden current-folder state with explicit Folder objects
This commit is contained in:
parent
a9a44ccb7e
commit
6fbd54c754
2 changed files with 59 additions and 46 deletions
|
|
@ -2,6 +2,7 @@ import os
|
|||
import json
|
||||
import shutil
|
||||
import tomllib
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from importlib import import_module
|
||||
from types import SimpleNamespace
|
||||
|
|
@ -44,6 +45,14 @@ def key(path) -> str:
|
|||
return Path(path).as_posix()
|
||||
|
||||
|
||||
@dataclass
|
||||
class Folder:
|
||||
"""The tag entries of one folder and the context file they are stored in"""
|
||||
path: Path
|
||||
file: Path
|
||||
entries: dict
|
||||
|
||||
|
||||
class OrvaniError(Exception):
|
||||
"""An error caused by the user's input, reported without a traceback"""
|
||||
|
||||
|
|
@ -80,10 +89,7 @@ class OrvaniTag:
|
|||
self.FILENAME = Path(path).name # folder context data filename ('.otag')
|
||||
self.CACHEFILE = cache # search index cache ('.otag.cache.json')
|
||||
self.SCHEMA = contexts # Default context schema
|
||||
self.DB = dict() # holds the folder context data
|
||||
self.FILE = '' # holds the folder context file path
|
||||
self.INDEX = self.__emptyindex() # our search index
|
||||
self.__cd('')
|
||||
|
||||
|
||||
def __emptyindex(self) -> dict:
|
||||
|
|
@ -92,15 +98,14 @@ class OrvaniTag:
|
|||
return index
|
||||
|
||||
|
||||
def __cd(self, path: str):
|
||||
def __load(self, path) -> Folder:
|
||||
"""
|
||||
Change folder context to other directory.
|
||||
Checks if a folder context file exists and loads context DB.
|
||||
Loads the entries of a folder from its context file, if it has one
|
||||
|
||||
path (str): path of the new folder context
|
||||
path (str): path of the folder
|
||||
"""
|
||||
self.FILE = Path(path) / self.FILENAME
|
||||
self.DB = self.__read(self.FILE) if self.FILE.exists() else dict()
|
||||
file = Path(path) / self.FILENAME
|
||||
return Folder(Path(path), file, self.__read(file) if file.exists() else dict())
|
||||
|
||||
|
||||
def config(self, folder: str='.') -> dict:
|
||||
|
|
@ -134,21 +139,21 @@ class OrvaniTag:
|
|||
return db
|
||||
|
||||
|
||||
def __save(self):
|
||||
def __save(self, folder: Folder):
|
||||
"""
|
||||
Save current folder context to file
|
||||
Saves the entries of a folder to its context file, deleting the file once it is empty
|
||||
"""
|
||||
file = Path(self.FILE)
|
||||
file = folder.file
|
||||
lines = []
|
||||
for name in sorted(self.DB):
|
||||
contexts = ' '.join(self.__collect_contexts(self.DB[name]))
|
||||
for name in sorted(folder.entries):
|
||||
contexts = ' '.join(self.__collect_contexts(folder.entries[name]))
|
||||
if contexts: # entries without any context are dropped
|
||||
slash = '/' if (file.parent / name).is_dir() else ''
|
||||
lines.append(f'{name}{slash}{SEPARATOR}{contexts}\n')
|
||||
|
||||
if not lines:
|
||||
file.unlink(missing_ok=True)
|
||||
return self.FILE
|
||||
return
|
||||
|
||||
# Swap in a finished temp file so a failed write never truncates the existing context.
|
||||
tmp = file.with_name(f'.{file.name}.{os.getpid()}.tmp')
|
||||
|
|
@ -159,7 +164,6 @@ class OrvaniTag:
|
|||
os.replace(tmp, file)
|
||||
finally:
|
||||
tmp.unlink(missing_ok=True)
|
||||
return self.FILE
|
||||
|
||||
|
||||
def __context(self, argv: list) -> dict:
|
||||
|
|
@ -251,8 +255,7 @@ class OrvaniTag:
|
|||
|
||||
folder (Path): path of folder
|
||||
"""
|
||||
self.__cd(folder)
|
||||
for file, context in self.DB.items():
|
||||
for file, context in self.__load(folder).entries.items():
|
||||
self.__updateindex(key(folder / file), context)
|
||||
|
||||
|
||||
|
|
@ -362,25 +365,26 @@ class OrvaniTag:
|
|||
if not target.exists(): # Only existing files and folders
|
||||
raise NotFound(f"File {file} doesn't exist.")
|
||||
|
||||
self.__cd(target.parent) # Load folder context
|
||||
folder = self.__load(target.parent) # Load folder context
|
||||
context = self.__context(argv) # Create context dict from argument list
|
||||
entry = folder.entries.get(target.name)
|
||||
|
||||
if target.name in self.DB: # Does the file exist in our local context DB?
|
||||
if entry: # Does the file exist in our local context DB?
|
||||
|
||||
# Add entries from argument context to context DB
|
||||
for context_key in self.SCHEMA.keys():
|
||||
if self.SCHEMA[context_key] != None:
|
||||
# Add to context and remove duplicate entries
|
||||
self.DB[target.name][context_key] = list(dict.fromkeys( self.DB[target.name][context_key] + context[context_key]))
|
||||
entry[context_key] = list(dict.fromkeys( entry[context_key] + context[context_key]))
|
||||
else:
|
||||
# Add the note
|
||||
if context_key in context:
|
||||
self.DB[target.name][context_key] = context[context_key]
|
||||
entry[context_key] = context[context_key]
|
||||
|
||||
else: # If file or folder does not exist in context DB yet, add it
|
||||
self.DB.update( { target.name : context } )
|
||||
folder.entries[target.name] = context
|
||||
|
||||
self.__save()
|
||||
self.__save(folder)
|
||||
return context
|
||||
|
||||
|
||||
|
|
@ -397,15 +401,16 @@ class OrvaniTag:
|
|||
if not target.exists(): # Only existing files and folders
|
||||
raise NotFound(f"File {file} doesn't exist.")
|
||||
|
||||
self.__cd(target.parent) # Load folder context
|
||||
folder = self.__load(target.parent) # Load folder context
|
||||
context = self.__context(argv) # Create context dict from argument list
|
||||
no_match = NoEntry(f"No match found for {file} and {' '.join(self.__collect_contexts(dict(context)))}.")
|
||||
if target.name not in self.DB: # Does the file exist in our local context DB?
|
||||
entry = folder.entries.get(target.name)
|
||||
if entry is None: # Does the file exist in our local context DB?
|
||||
raise no_match
|
||||
|
||||
if not argv:
|
||||
del self.DB[target.name]
|
||||
self.__save()
|
||||
del folder.entries[target.name]
|
||||
self.__save(folder)
|
||||
return None
|
||||
|
||||
removed = False
|
||||
|
|
@ -413,12 +418,12 @@ class OrvaniTag:
|
|||
# Only works for context list items
|
||||
if self.SCHEMA[context_key] != None:
|
||||
for item in context[context_key]:
|
||||
if item in self.DB[target.name][context_key]:
|
||||
self.DB[target.name][context_key].remove(item)
|
||||
if item in entry[context_key]:
|
||||
entry[context_key].remove(item)
|
||||
removed = True
|
||||
if not removed:
|
||||
raise no_match
|
||||
self.__save()
|
||||
self.__save(folder)
|
||||
return context
|
||||
|
||||
|
||||
|
|
@ -434,14 +439,14 @@ class OrvaniTag:
|
|||
target = Path(file)
|
||||
if not target.exists(): # Does the target file exist?
|
||||
raise NotFound(f"File {file} doesn't exist.")
|
||||
self.__cd(target.parent) # Load its local context
|
||||
if target.name not in self.DB: # Do we have a record in the DB?
|
||||
folder = self.__load(target.parent) # Load its local context
|
||||
if target.name not in folder.entries: # Do we have a record in the DB?
|
||||
raise NoEntry(f"No entry for {file} found.")
|
||||
del self.DB[target.name] # Delete it
|
||||
self.__save() # deletes the context file once it is empty
|
||||
del folder.entries[target.name] # Delete it
|
||||
self.__save(folder) # deletes the context file once it is empty
|
||||
if fop: # delete the referenced file too?
|
||||
target.unlink(missing_ok=True)
|
||||
return None if self.FILE.exists() else self.FILE
|
||||
return None if folder.file.exists() else folder.file
|
||||
|
||||
|
||||
def move(self, source: str, dest: str, retain: bool=False, fop: bool=True):
|
||||
|
|
@ -469,14 +474,13 @@ class OrvaniTag:
|
|||
if not target.parent.is_dir():
|
||||
raise NotFound(f"Destination {dest} doesn't exist.")
|
||||
|
||||
self.__cd(src.parent) # load local context
|
||||
sourcedata = self.DB.get(src.name) # load its context data
|
||||
sourcedata = self.__load(src.parent).entries.get(src.name) # load its context data
|
||||
if sourcedata is None and not fop: # without an entry there is nothing to move
|
||||
raise NoEntry(f"No entry for {source} found.")
|
||||
if sourcedata is not None:
|
||||
self.__cd(target.parent) # load local context of destination folder
|
||||
self.DB.update( { target.name : sourcedata } ) # update its DB with the context data of our source file
|
||||
self.__save() # and save it
|
||||
destination = self.__load(target.parent) # load local context of destination folder
|
||||
destination.entries[target.name] = sourcedata # with the context data of our source file
|
||||
self.__save(destination) # and save it
|
||||
if not retain: # do we copy or move the data?
|
||||
self.delete(source) # delete for move
|
||||
if fop: # do we also physically move or copy the reference file?
|
||||
|
|
@ -504,13 +508,13 @@ class OrvaniTag:
|
|||
for folder, files in walk(target, recursive):
|
||||
if self.FILENAME not in files:
|
||||
continue
|
||||
self.__cd(folder) # load folder context
|
||||
loaded = self.__load(folder) # load folder context
|
||||
# delete the entries of files that do not exist anymore
|
||||
missing = [file for file in self.DB if not (folder / file).exists()]
|
||||
missing = [file for file in loaded.entries if not (folder / file).exists()]
|
||||
for file in missing:
|
||||
del self.DB[file]
|
||||
del loaded.entries[file]
|
||||
if missing:
|
||||
self.__save() # deletes the context file once it is empty
|
||||
self.__save(loaded) # deletes the context file once it is empty
|
||||
count += len(missing)
|
||||
return count
|
||||
|
||||
|
|
|
|||
|
|
@ -207,7 +207,7 @@ def test_remove_saves_once(tree, monkeypatch):
|
|||
o.add('a.mp3', '#lofi', '#snare', '#keep')
|
||||
saves = []
|
||||
save = o._OrvaniTag__save
|
||||
monkeypatch.setattr(o, '_OrvaniTag__save', lambda: saves.append(1) or save())
|
||||
monkeypatch.setattr(o, '_OrvaniTag__save', lambda folder: saves.append(1) or save(folder))
|
||||
o.remove('a.mp3', '#lofi', '#snare')
|
||||
assert len(saves) == 1
|
||||
assert lines(tree) == ['a.mp3 --- #keep']
|
||||
|
|
@ -294,3 +294,12 @@ def test_clean_with_a_file_path_cleans_its_folder(tree):
|
|||
(tree / 'b.mp3').unlink()
|
||||
assert o.clean('a.mp3', False) == 1
|
||||
assert lines(tree) == ['a.mp3 --- #lofi']
|
||||
|
||||
|
||||
def test_broken_context_file_only_affects_its_own_folder(tree):
|
||||
(tree / orvani.CONTEXT_FILE).write_text('broken line\n')
|
||||
o = OrvaniTag()
|
||||
o.add('sub/d.txt', '#doc')
|
||||
assert lines(tree / 'sub') == ['d.txt --- #doc']
|
||||
with pytest.raises(FormatError):
|
||||
o.add('a.mp3', '#lofi')
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue