89 lines
2.1 KiB
Python
89 lines
2.1 KiB
Python
import re
|
|
from dataclasses import dataclass
|
|
from typing import Iterator, Protocol
|
|
|
|
from fastapi import HTTPException, status
|
|
from webdav3.exceptions import RemoteResourceNotFound
|
|
|
|
from ..dav_common import caldav_principal, webdav_list
|
|
|
|
|
|
@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 CalendarNameLister:
|
|
async def __call__(self) -> Iterator[str]:
|
|
return (
|
|
cal.name
|
|
for cal in caldav_principal().calendars()
|
|
)
|
|
|
|
|
|
@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 PrefixUnique:
|
|
finder: PrefixFinder
|
|
|
|
@property
|
|
def responses(self) -> dict:
|
|
return {
|
|
status.HTTP_200_OK: {
|
|
"description": "Operation successful",
|
|
},
|
|
status.HTTP_404_NOT_FOUND: {
|
|
"description": "Prefix not found",
|
|
"content": None,
|
|
},
|
|
status.HTTP_409_CONFLICT: {
|
|
"description": "Ambiguous prefix",
|
|
"content": None,
|
|
},
|
|
}
|
|
|
|
async def __call__(self, prefix: str) -> str:
|
|
names = await self.finder(prefix)
|
|
|
|
try:
|
|
name = next(names)
|
|
|
|
except StopIteration:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
|
|
|
|
if any(True for _ in names):
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT)
|
|
|
|
return name
|