80 lines
2 KiB
Python
80 lines
2 KiB
Python
import re
|
|
from dataclasses import dataclass
|
|
from typing import Iterator, Protocol
|
|
|
|
from fastapi import HTTPException, status
|
|
from webdav3.exceptions import RemoteResourceNotFound
|
|
|
|
from .. import webdav_list
|
|
from ..dav_file import DavFile
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class NameLister(Protocol):
|
|
def __call__(self) -> Iterator[str]:
|
|
...
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FileNameLister:
|
|
remote_path: str
|
|
re: re.Pattern[str]
|
|
|
|
async def __call__(self) -> Iterator[str]:
|
|
try:
|
|
file_names = await webdav_list(self.remote_path)
|
|
|
|
return (
|
|
name
|
|
for name in file_names
|
|
if self.re.search(name)
|
|
)
|
|
|
|
except RemoteResourceNotFound:
|
|
return iter(())
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PrefixFinder:
|
|
lister: NameLister
|
|
|
|
async def __call__(self, prefix: str) -> Iterator[str]:
|
|
return (
|
|
file_name
|
|
for file_name in (await self.lister())
|
|
if file_name.lower().startswith(prefix.lower())
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FilePrefixLoader:
|
|
finder: PrefixFinder
|
|
|
|
@property
|
|
def responses(self) -> dict:
|
|
return {
|
|
status.HTTP_200_OK: {
|
|
"description": "Operation successful",
|
|
},
|
|
status.HTTP_404_NOT_FOUND: {
|
|
"description": "file not found in " +
|
|
repr(self.finder.lister.remote_path),
|
|
"content": None,
|
|
},
|
|
status.HTTP_409_CONFLICT: {
|
|
"description": "ambiguous file name for " +
|
|
repr(self.finder.lister.remote_path),
|
|
"content": None,
|
|
},
|
|
}
|
|
|
|
async def __call__(self, prefix: str) -> DavFile:
|
|
file_names = list(await self.finder(prefix))
|
|
|
|
if not (file_names):
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
|
|
|
|
elif len(file_names) > 1:
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT)
|
|
|
|
return DavFile(f"{self.finder.lister.remote_path}/{file_names[0]}")
|