feat(backend): use Path object from pathlib instead of os.path library

Author: kolomenkin

config.py

@@ -3,6 +3,7 @@
 # Licensed under MIT (https://github.com/kolomenkin/limbo/blob/master/LICENSE)
 #
 import os
+from pathlib import Path
 
 
 def read_env(name: str, default_value: str) -> str:
@@ -16,7 +17,7 @@ def read_env(name: str, default_value: str) -> str:
 LISTEN_HOST = read_env('LIMBO_LISTEN_HOST', '127.0.0.1')
 LISTEN_PORT = int(read_env('LIMBO_LISTEN_PORT', '8080'))
 
-STORAGE_DIRECTORY = read_env('LIMBO_STORAGE_DIRECTORY', './storage')
+STORAGE_DIRECTORY = Path(read_env('LIMBO_STORAGE_DIRECTORY', './storage'))
 
 # STORAGE_WEB_URL_BASE allows to specify alternative web url to
 # read files stored in STORAGE_DIRECTORY through HTTP/HTTPS.

lib_common.py

@@ -2,9 +2,8 @@
 # Copyright 2018-2022 Sergey Kolomenkin
 # Licensed under MIT (https://github.com/kolomenkin/limbo/blob/master/LICENSE)
 #
-import os
-import stat
+from pathlib import Path
 
 
-def get_file_modified_unixtime(pathname: str) -> float:
-    return os.stat(pathname)[stat.ST_MTIME]
+def get_file_modified_unixtime(pathname: Path) -> float:
+    return pathname.stat().st_mtime

lib_file_storage.py

@@ -2,12 +2,11 @@
 # Copyright 2018-2022 Sergey Kolomenkin
 # Licensed under MIT (https://github.com/kolomenkin/limbo/blob/master/LICENSE)
 #
-import io
 import logging
-import os
 import re
 import threading
 from dataclasses import dataclass
+from pathlib import Path
 from time import sleep, time
 from typing import Any, List, Optional, Sequence
 from uuid import uuid4
@@ -44,19 +43,19 @@ def clean_filename(filename: str) -> str:
 
 
 class AtomicFile:
-    def __init__(self, temp_filename: str, final_filename: str):
-        if os.path.isfile(final_filename):
+    def __init__(self, temp_filename: Path, final_filename: Path):
+        if final_filename.is_file():
             raise Exception('Destination file already exists')
-        self._temp_filename = temp_filename
-        self._final_filename = final_filename
-        self._fd = io.open(self._temp_filename, 'wb')  # pylint: disable=consider-using-with
+        self._temp_filename: Path = temp_filename
+        self._final_filename: Path = final_filename
+        self._fd = self._temp_filename.open('wb')  # pylint: disable=consider-using-with
 
     def write(self, data: bytes) -> None:
         self._fd.write(data)
 
     def close(self) -> None:
         self._fd.close()
-        os.rename(self._temp_filename, self._final_filename)
+        self._temp_filename.rename(self._final_filename)
 
     def __enter__(self) -> 'AtomicFile':
         return self
@@ -65,28 +64,28 @@ def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
         self._fd.close()
         if exc_tb is None:
             # No exception, so rename
-            os.rename(self._temp_filename, self._final_filename)
+            self._temp_filename.rename(self._final_filename)
 
 
 @dataclass
 class DisplayFileItem:
-    full_disk_filename: str
+    full_disk_filename: Path
     url_filename: str
     display_filename: str
 
 
 @dataclass
 class StorageFileItem:
-    storage_directory: str
+    storage_directory: Path
     disk_filename: str
     display_filename: str
 
 
 class FileStorage:
-    def __init__(self, storage_directory: str, max_store_time_seconds: int):
+    def __init__(self, storage_directory: Path, max_store_time_seconds: int):
         LOGGER.info('FileStorage: create("%s", max %d sec)', storage_directory, max_store_time_seconds)
-        self._storage_directory = os.path.abspath(storage_directory)
-        self._temp_directory = os.path.join(self._storage_directory, 'incomplete')
+        self._storage_directory: Path = storage_directory.absolute()
+        self._temp_directory: Path = self._storage_directory / 'incomplete'
         self._max_store_time_seconds = max_store_time_seconds
         self._retention_thread: Optional[threading.Thread] = None
         self._protect_stop = threading.Lock()
@@ -109,16 +108,15 @@ def stop(self) -> None:
             self._retention_thread.join()
 
     def enumerate_files(self) -> Sequence[DisplayFileItem]:
-        if not os.path.isdir(self._storage_directory):
+        if not self._storage_directory.is_dir():
             return []
         files: List[DisplayFileItem] = []
-        for disk_filename in os.listdir(self._storage_directory):
-            fullname = os.path.join(self._storage_directory, disk_filename)
-            if os.path.isfile(fullname):
-                url_filename = self._fname_disk_to_url(disk_filename)
-                display_filename = self._fname_disk_to_display(disk_filename)
+        for disk_filename in self._storage_directory.iterdir():
+            if disk_filename.is_file():
+                url_filename = self._fname_disk_to_url(disk_filename.name)
+                display_filename = self._fname_disk_to_display(disk_filename.name)
                 files.append(DisplayFileItem(
-                    full_disk_filename=fullname,
+                    full_disk_filename=disk_filename,
                     url_filename=url_filename,
                     display_filename=display_filename,
                 ))
@@ -128,8 +126,8 @@ def open_file_writer(self, original_filename: str) -> AtomicFile:
         self._create_dirs()
         disk_filename = self._fname_original_to_disk(original_filename)
         temp_disk_filename = f'{uuid4().hex}.{disk_filename}'
-        temp_fullname = os.path.join(self._temp_directory, temp_disk_filename)
-        fullname = os.path.join(self._storage_directory, disk_filename)
+        temp_fullname = self._temp_directory / temp_disk_filename
+        fullname = self._storage_directory / disk_filename
         LOGGER.info('FileStorage: Upload file: %s', disk_filename)
         return AtomicFile(temp_fullname, fullname)
 
@@ -144,28 +142,26 @@ def get_file_info_to_read(self, url_filename: str) -> StorageFileItem:
 
     def remove_file(self, url_filename: str) -> None:
         disk_filename = self._fname_url_to_disk(url_filename)
-        fullname = os.path.join(self._storage_directory, disk_filename)
-        file_size = os.path.getsize(fullname)
+        fullname = self._storage_directory / disk_filename
+        file_size = fullname.stat().st_size
         LOGGER.info('FileStorage: Remove file: "%s"; size: %d', disk_filename, file_size)
-        os.remove(fullname)
+        fullname.unlink()
 
     def remove_all_files(self) -> None:
-        if not os.path.isdir(self._storage_directory):
+        if not self._storage_directory.is_dir():
             return
-        for disk_filename in os.listdir(self._storage_directory):
-            fullname = os.path.join(self._storage_directory, disk_filename)
-            if os.path.isfile(fullname):
-                file_size = os.path.getsize(fullname)
-                LOGGER.info('FileStorage: Remove file: "%s"; size: %d', disk_filename, file_size)
-                os.remove(fullname)
-        if not os.path.isdir(self._temp_directory):
+        for disk_filename in self._storage_directory.iterdir():
+            if disk_filename.is_file():
+                file_size = disk_filename.stat().st_size
+                LOGGER.info('FileStorage: Remove file: "%s"; size: %d', disk_filename.name, file_size)
+                disk_filename.unlink()
+        if not self._temp_directory.is_dir():
             return
-        for disk_filename in os.listdir(self._temp_directory):
-            fullname = os.path.join(self._temp_directory, disk_filename)
-            if os.path.isfile(fullname):
-                file_size = os.path.getsize(fullname)
-                LOGGER.info('FileStorage: Remove temp file: "%s"; size: %d', disk_filename, file_size)
-                os.remove(fullname)
+        for disk_filename in self._temp_directory.iterdir():
+            if disk_filename.is_file():
+                file_size = disk_filename.stat().st_size
+                LOGGER.info('FileStorage: Remove temp file: "%s"; size: %d', disk_filename.name, file_size)
+                disk_filename.unlink()
 
     @classmethod
     def _fname_original_to_disk(cls, original_filename: str) -> str:
@@ -191,11 +187,11 @@ def _canonize_file(filename: str) -> str:
         return canonized
 
     def _create_dirs(self) -> None:
-        if not os.path.isdir(self._storage_directory):
-            os.makedirs(self._storage_directory, 0o755)
+        if not self._storage_directory.is_dir():
+            self._storage_directory.mkdir(mode=0o755)
 
-        if not os.path.isdir(self._temp_directory):
-            os.makedirs(self._temp_directory, 0o755)
+        if not self._temp_directory.is_dir():
+            self._temp_directory.mkdir(mode=0o755)
 
     def _retention_thread_procedure(self) -> None:
         LOGGER.info('FileStorage: Retention thread started')
@@ -222,24 +218,22 @@ def _retention_thread_procedure(self) -> None:
 
     def _check_retention(self) -> None:
         now = time()
-        if not os.path.isdir(self._storage_directory):
+        if not self._storage_directory.is_dir():
             return
-        for file in os.listdir(self._storage_directory):
-            fullname = os.path.join(self._storage_directory, file)
-            if os.path.isfile(fullname):
-                modified_unixtime = get_file_modified_unixtime(fullname)
+        for file in self._storage_directory.iterdir():
+            if file.is_file():
+                modified_unixtime = get_file_modified_unixtime(file)
                 if now - modified_unixtime > self._max_store_time_seconds:
-                    file_size = os.path.getsize(fullname)
-                    LOGGER.info('FileStorage: Remove outdated file: "%s"; size: %d', fullname, file_size)
-                    os.remove(fullname)
+                    file_size = file.stat().st_size
+                    LOGGER.info('FileStorage: Remove outdated file: "%s"; size: %d', file, file_size)
+                    file.unlink()
 
-        if not os.path.isdir(self._temp_directory):
+        if not self._temp_directory.is_dir():
             return
-        for file in os.listdir(self._temp_directory):
-            fullname = os.path.join(self._temp_directory, file)
-            if os.path.isfile(fullname):
-                modified_unixtime = get_file_modified_unixtime(fullname)
+        for file in self._temp_directory.iterdir():
+            if file.is_file():
+                modified_unixtime = get_file_modified_unixtime(file)
                 if now - modified_unixtime > 15 * 60:  # every 15 minutes
-                    file_size = os.path.getsize(fullname)
-                    LOGGER.info('FileStorage: Remove outdated temp file: "%s"; size: %d', fullname, file_size)
-                    os.remove(fullname)
+                    file_size = file.stat().st_size
+                    LOGGER.info('FileStorage: Remove outdated temp file: "%s"; size: %d', file, file_size)
+                    file.unlink()

server.py

@@ -12,6 +12,7 @@
 import sys
 import threading
 import urllib.parse
+from pathlib import Path
 from time import time
 from typing import Any, Dict, List, Mapping, Optional, Union
 
@@ -93,7 +94,7 @@ def root_page() -> ViewResponse:
                 'display_filename': file.display_filename,
                 'url': URLPREFIX + urllib.parse.quote(file.url_filename),
                 'url_filename': file.url_filename,
-                'size': format_size(os.path.getsize(file.full_disk_filename)),
+                'size': format_size(file.full_disk_filename.stat().st_size),
                 'age': format_age(int(now - modified_unixtime)),
                 'sortBy': now - modified_unixtime,
             }
@@ -120,7 +121,7 @@ def cgi_enumerate() -> MethodResponse:
                 'display_filename': file.display_filename,
                 'url': URLPREFIX + urllib.parse.quote(file.url_filename),
                 'url_filename': file.url_filename,
-                'size': os.path.getsize(file.full_disk_filename),
+                'size': file.full_disk_filename.stat().st_size,
                 'modified': modified_unixtime,
             }
         )
@@ -232,8 +233,8 @@ def cgi_remove_all() -> MethodResponse:
 @bottle_route('/static/<urlpath:path>')
 def server_static(urlpath: str) -> RouteResponse:
     LOGGER.debug('Static file requested: %s', urlpath)
-    root_folder = os.path.abspath(os.path.dirname(__file__))
-    response = bottle.static_file(urlpath, root=os.path.join(root_folder, 'static'))
+    root_folder = Path(__file__).parent.absolute()
+    response = bottle.static_file(urlpath, root=root_folder / 'static')
     response.set_header('Cache-Control', 'public, max-age=604800')
     return response
 
@@ -348,7 +349,7 @@ def main() -> None:
 
     LOGGER.info('Start server...')
 
-    bottle.TEMPLATE_PATH = [os.path.join(os.path.dirname(__file__), 'static', 'templates')]
+    bottle.TEMPLATE_PATH = [Path(__file__).parent / 'static' / 'templates']
 
     server_thread = threading.Thread(target=run_bottle, daemon=True, name='bottle')
     server_thread.start()

tests/test_file_storage.py

@@ -1,16 +1,12 @@
 from base64 import b64decode
 from dataclasses import dataclass
-from os import path as os_path
+from pathlib import Path
 from tempfile import TemporaryDirectory
 from typing import Sequence
 from unittest import TestCase
 
 from numpy import random
 
-# add parent dir to search for imported modules
-# import os, sys
-# script_dir = os.path.dirname(os.path.abspath(__file__))
-# sys.path.insert(0, script_dir + '/../')
 from lib_file_storage import DisplayFileItem, FileStorage, StorageFileItem
 
 
@@ -30,7 +26,7 @@ class TempStorage:
 def get_temp_file_storage() -> TempStorage:
     temp_directory = TemporaryDirectory()  # pylint: disable=consider-using-with
     print('created temporary directory: ' + temp_directory.name)
-    storage = FileStorage(temp_directory.name, 24 * 3600)
+    storage = FileStorage(Path(temp_directory.name), 24 * 3600)
     return TempStorage(temp_directory=temp_directory, storage=storage)
 
 
@@ -53,14 +49,14 @@ def process_single_file(self, original_filename: str, original_filedata: bytes)
         url_filename = item.url_filename
         display_filename = item.display_filename
         self.assertEqual(display_filename, original_filename)
-        size = os_path.getsize(item.full_disk_filename)
+        size = item.full_disk_filename.stat().st_size
         self.assertEqual(size, len(original_filedata))
 
         info: StorageFileItem = storage.get_file_info_to_read(url_filename)
-        self.assertEqual(info.storage_directory, temp_storage.temp_directory.name)
+        self.assertEqual(info.storage_directory, Path(temp_storage.temp_directory.name))
         self.assertEqual(info.display_filename, display_filename)
 
-        fullpath = os_path.join(info.storage_directory, info.disk_filename)
+        fullpath = info.storage_directory / info.disk_filename
         with open(fullpath, 'rb') as file:
             filedata = file.read()
             self.assertEqual(filedata, original_filedata)
@@ -122,13 +118,13 @@ def test_multiple_files(self) -> None:
         item = files[0]
         url_filename1 = item.url_filename
         self.assertEqual(item.display_filename, original_filename1)
-        size = os_path.getsize(item.full_disk_filename)
+        size = item.full_disk_filename.stat().st_size
         self.assertEqual(size, len(original_filedata1))
 
         item = files[1]
         url_filename2 = item.url_filename
         self.assertEqual(item.display_filename, original_filename2)
-        size = os_path.getsize(item.full_disk_filename)
+        size = item.full_disk_filename.stat().st_size
         self.assertEqual(size, len(original_filedata2))
 
         storage.remove_file(url_filename1)

tests/test_server.py

@@ -4,6 +4,7 @@
 from base64 import b64decode
 from dataclasses import dataclass
 from datetime import datetime
+from pathlib import Path
 from tempfile import TemporaryDirectory
 from typing import Any, List, Optional, Sequence
 from unittest import TestCase
@@ -51,9 +52,9 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
 
     @staticmethod
     def run_child_server(server_name: str, host: str, port: int) -> RunningServer:
-        script_dir = os.path.dirname(os.path.abspath(__file__))
-        root_dir = os.path.join(script_dir, '..')
-        server_py = os.path.join(root_dir, 'server.py')
+        script_dir = Path(__file__).parent.absolute()
+        root_dir = script_dir.parent
+        server_py = root_dir / 'server.py'
 
         temp_directory = TemporaryDirectory()  # pylint: disable=consider-using-with
         log('created temporary directory: ' + temp_directory.name)

utils/speedtest.py

@@ -4,6 +4,7 @@
 import sys
 from dataclasses import dataclass
 from datetime import datetime
+from pathlib import Path
 from tempfile import TemporaryDirectory
 from typing import List, Optional, Sequence
 
@@ -43,9 +44,9 @@ def __init__(self) -> None:
 
     @staticmethod
     def run_child_server(server_name: str, port: int) -> RunningServer:
-        script_dir = os.path.dirname(os.path.abspath(__file__))
-        root_dir = os.path.join(script_dir, '..')
-        server_py = os.path.join(root_dir, 'server.py')
+        script_dir = Path(__file__).parent.absolute()
+        root_dir = script_dir.parent
+        server_py = root_dir / 'server.py'
 
         temp_directory = TemporaryDirectory()  # pylint: disable=consider-using-with
         LOGGER.info('created temporary directory: %s', temp_directory.name)