Refactor file paths and URLs handling

control/deciphonctl/file_path.py

@@ -0,0 +1,6 @@
+from deciphon_core.schema import FilePath
+from pydantic import TypeAdapter
+
+
+def file_path(x: FilePath):
+    return TypeAdapter(FilePath).validate_python(x)

control/deciphonctl/models.py

@@ -54,8 +54,7 @@ def hmm_file_name(self):
         return HMMFileName(name=str(Path(self.name).with_suffix(".hmm")))
 
 
-class HMMFile(HMMName):
-    ...
+class HMMFile(HMMName): ...
 
 
 class DBFile(DBName):

control/deciphonctl/presser.py

@@ -5,10 +5,11 @@
 from deciphon_core.press import PressContext
 from deciphon_core.schema import HMMFile
 from loguru import logger
-from pydantic import FilePath, HttpUrl
+from pydantic import HttpUrl
 
 from deciphonctl.consumer import Consumer
 from deciphonctl.download import download
+from deciphonctl.file_path import file_path
 from deciphonctl.files import atomic_file_creation, remove_temporary_files
 from deciphonctl.models import DBFile, JobUpdate, PressRequest
 from deciphonctl.permissions import normalise_file_permissions
@@ -39,7 +40,7 @@ def _hmm2dcp(self, url: HttpUrl, hmmfile: Path, request: PressRequest):
 
     def _press(self, hmmfile: Path, req: PressRequest):
         dcpfile = hmmfile.with_suffix(".dcp")
-        hmm = HMMFile(path=FilePath(hmmfile))
+        hmm = HMMFile(path=file_path(hmmfile))
         db = req.db
         with PressContext(hmm, gencode=db.gencode, epsilon=db.epsilon) as press:
             self._qout.put(JobUpdate.run(req.job_id, 0).model_dump_json())

control/deciphonctl/scanner.py

@@ -7,10 +7,10 @@
 from deciphon_core.schema import DBFile, HMMFile, NewSnapFile
 from deciphon_core.sequence import Sequence
 from loguru import logger
-from pydantic import FilePath
 
 from deciphonctl.consumer import Consumer
 from deciphonctl.download import download
+from deciphonctl.file_path import file_path
 from deciphonctl.files import (
     atomic_file_creation,
     remove_temporary_files,
@@ -51,10 +51,10 @@ def callback(self, message: str):
         with unique_temporary_file(".dcs") as t:
             snap = NewSnapFile(path=t)
 
-            db = DBFile(path=FilePath(dbfile))
+            db = DBFile(path=file_path(dbfile))
 
             logger.info("starting h3daemon")
-            with H3Daemon(HMMFile(path=FilePath(hmmfile)), stdout=DEVNULL) as daemon:
+            with H3Daemon(HMMFile(path=file_path(hmmfile)), stdout=DEVNULL) as daemon:
                 params = Params(
                     num_threads=self._num_threads,
                     multi_hits=x.multi_hits,

control/deciphonctl/sched.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-from requests_toolbelt import MultipartEncoder
 import urllib.parse
 from pathlib import Path
 from typing import Any, Optional
@@ -10,8 +9,10 @@
 from loguru import logger
 from pydantic import BaseModel, FilePath, HttpUrl
 from requests.models import HTTPError
+from requests_toolbelt import MultipartEncoder
 
 from deciphonctl.models import DBFile, HMMFile, JobUpdate, Scan
+from deciphonctl.url import http_url
 
 
 class SchedHTTPError(HTTPError):
@@ -121,7 +122,9 @@ def seq_list(self):
         return self.get(self.url("seqs")).json()
 
     def snap_post(self, scan_id: int, snap: FilePath):
-        post = UploadPost(url=HttpUrl(self.url(f"scans/{scan_id}/snap.dcs")), fields={})
+        post = UploadPost(
+            url=http_url(self.url(f"scans/{scan_id}/snap.dcs")), fields={}
+        )
         self.upload(snap, post)
 
     def snap_get(self, scan_id: int):
@@ -147,20 +150,20 @@ def _request(self, path: str):
 
     def download_hmm_url(self, filename: str):
         x = self._request(f"hmms/presigned-download/{filename}")
-        return HttpUrl(x["url"])
+        return http_url(x["url"])
 
     def download_db_url(self, filename: str):
         x = self._request(f"dbs/presigned-download/{filename}")
-        return HttpUrl(x["url"])
+        return http_url(x["url"])
 
     def upload_hmm_post(self, filename: str):
         x = self._request(f"hmms/presigned-upload/{filename}")
-        url = self._sched.s3_url if self._sched.s3_url else HttpUrl(x["url"])
+        url = self._sched.s3_url if self._sched.s3_url else http_url(x["url"])
         return UploadPost(url=url, fields=x["fields"])
 
     def upload_db_post(self, filename: str):
         x = self._request(f"dbs/presigned-upload/{filename}")
-        url = self._sched.s3_url if self._sched.s3_url else HttpUrl(x["url"])
+        url = self._sched.s3_url if self._sched.s3_url else http_url(x["url"])
         return UploadPost(url=url, fields=x["fields"])
 
 

control/deciphonctl/settings.py

@@ -3,11 +3,13 @@
 from pydantic import HttpUrl
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
+from deciphonctl.url import http_url
+
 
 class Settings(BaseSettings):
     model_config = SettingsConfigDict(env_prefix="deciphonctl_")
 
-    sched_url: HttpUrl = HttpUrl("http://localhost")
+    sched_url: HttpUrl = http_url("http://localhost")
 
     s3_url: Optional[HttpUrl] = None
     s3_bucket: str = "deciphon"

control/deciphonctl/url.py

@@ -1,13 +1,11 @@
-import urllib.parse
-from deciphonctl import settings
-from pydantic import HttpUrl
-
-
-def url(endpoint: str):
-    return urllib.parse.urljoin(settings.sched_url.unicode_string(), endpoint)
+from pydantic import HttpUrl, TypeAdapter
 
 
 def url_filename(url: HttpUrl):
     path = url.path
     assert isinstance(path, str)
     return path.split("/")[-1]
+
+
+def http_url(url: str):
+    return TypeAdapter(HttpUrl).validate_strings(url)

control/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "deciphonctl"
-version = "0.4.1"
+version = "0.4.2"
 description = "Control Deciphon server."
 authors = ["Danilo Horta <[email protected]>"]
 license = "MIT"