aboutsummaryrefslogtreecommitdiff
path: root/mu4web/mu.py
diff options
context:
space:
mode:
Diffstat (limited to 'mu4web/mu.py')
-rw-r--r--mu4web/mu.py70
1 files changed, 48 insertions, 22 deletions
diff --git a/mu4web/mu.py b/mu4web/mu.py
index d75d7a9..8efbd38 100644
--- a/mu4web/mu.py
+++ b/mu4web/mu.py
@@ -11,17 +11,20 @@ import xml.dom.minidom
import xml.dom
from xdg.BaseDirectory import xdg_cache_home
import os.path
-import xapian
+from . import xapian
from datetime import datetime
+from os import PathLike
+from pathlib import Path
from typing import (
Optional,
+ TypedDict,
)
parser = BytesParser(policy=email.policy.default)
-def find_file(id: str) -> Optional[str]:
+def find_file(id: str) -> Optional[PathLike]:
cmd = subprocess.run(['mu', 'find', '-u', f'i:{id}',
'--fields', 'l'],
stdout=PIPE)
@@ -31,7 +34,7 @@ def find_file(id: str) -> Optional[str]:
return None
if cmd.returncode != 0:
raise MuError(cmd.returncode)
- return filename
+ return Path(filename)
def get_mail(id: str) -> email.message.Message:
@@ -41,7 +44,11 @@ def get_mail(id: str) -> email.message.Message:
[Raises]
MuError
"""
- with open(find_file(id), "rb") as f:
+ filename = find_file(id)
+ if not filename:
+ # TODO better error here
+ raise MuError(1)
+ with open(filename, "rb") as f:
mail = parser.parse(f)
return mail
@@ -57,10 +64,10 @@ class MuError(Exception):
self.returncode: int = returncode
self.msg: str = MuError.codes.get(returncode, 'Unknown Error')
- def __repr__(self):
+ def __repr__(self) -> str:
return f'MuError({self.returncode}, "{self.msg}")'
- def __str__(self):
+ def __str__(self) -> str:
return repr(self)
@@ -118,7 +125,7 @@ def search(query: str,
return message_list
-def base_directory():
+def base_directory() -> PathLike:
"""
Returns where where mu stores its files.
@@ -127,28 +134,47 @@ def base_directory():
TODO make this configurable.
"""
- return os.getenv('MUHOME') or os.path.join(xdg_cache_home, 'mu')
+ return Path(os.getenv('MUHOME') or os.path.join(xdg_cache_home, 'mu'))
-def info():
+class MuInfo(TypedDict):
+ database_path: str
+ changed: datetime
+ created: datetime
+ indexed: datetime
+ maildir: str
+ schema_version: str
+ messages_in_store: int
+
+
+def info() -> MuInfo:
db = os.path.join(base_directory(), "xapian")
- info = {
- 'database-path': db,
- }
+ def f(key: str) -> datetime:
+ return datetime.fromtimestamp(int(xapian.metadata_get(db, key), 16))
- for key in ['changed', 'created', 'indexed']:
- info[key] = datetime.fromtimestamp(int(xapian.metadata_get(db, key), 16))
+ changed = f('changed')
+ created = f('created')
+ indexed = f('indexed')
- for key in ['maildir', 'schema-version']:
- info[key] = xapian.metadata_get(db, key)
+ maildir = xapian.metadata_get(db, 'maildir')
+ schema_version = xapian.metadata_get(db, 'schema-version')
- cmd = subprocess.Popen(['xapian-delve', '-V3', '-1', db], stdout=subprocess.PIPE)
+ cmd = subprocess.Popen(['xapian-delve', '-V3', '-1', db],
+ stdout=subprocess.PIPE)
# Start at minus one to ignore header
count = -1
- for line in cmd.stdout:
- count += 1
- info['messages-in-store'] = count
-
- return info
+ if cmd.stdout:
+ for line in cmd.stdout:
+ count += 1
+
+ return {
+ 'database_path': db,
+ 'messages_in_store': count,
+ 'changed': changed,
+ 'created': created,
+ 'indexed': indexed,
+ 'maildir': maildir,
+ 'schema_version': schema_version,
+ }