145 lines
3.4 KiB
Python
145 lines
3.4 KiB
Python
"""
|
|
Dependables for defining Routers.
|
|
"""
|
|
|
|
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_list, webdav_list
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class NameLister(Protocol):
|
|
"""
|
|
Can be called to create an iterator containing some names.
|
|
"""
|
|
|
|
async def __call__(self) -> Iterator[str]:
|
|
...
|
|
|
|
|
|
_RESPONSE_OK = {
|
|
status.HTTP_200_OK: {
|
|
"description": "Operation successful",
|
|
},
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FileNameLister:
|
|
"""
|
|
Can be called to create an iterator containing file names.
|
|
|
|
File names listed will be in `remote_path` and will match the RegEx `re`.
|
|
"""
|
|
|
|
remote_path: str
|
|
re: re.Pattern[str]
|
|
|
|
@property
|
|
def responses(self) -> dict:
|
|
return {
|
|
**_RESPONSE_OK,
|
|
status.HTTP_404_NOT_FOUND: {
|
|
"description": f"{self.remote_path!r} not found",
|
|
"content": None,
|
|
},
|
|
}
|
|
|
|
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:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CalendarNameLister:
|
|
"""
|
|
Can be called to create an iterator containing calendar names.
|
|
"""
|
|
|
|
async def __call__(self) -> Iterator[str]:
|
|
return await caldav_list()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PrefixFinder:
|
|
"""
|
|
Can be called to create an iterator containing some names, all starting
|
|
with a given prefix.
|
|
|
|
All names will be taken from the list produced by the called `lister`.
|
|
"""
|
|
|
|
lister: NameLister
|
|
|
|
@property
|
|
def responses(self) -> dict:
|
|
return {
|
|
**_RESPONSE_OK,
|
|
status.HTTP_404_NOT_FOUND: {
|
|
"description": "Failure in lister " +
|
|
repr(self.lister.__class__.__name__),
|
|
"content": None,
|
|
},
|
|
}
|
|
|
|
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:
|
|
"""
|
|
Can be called to determine if a given prefix is unique in the list
|
|
produced by the called `finder`.
|
|
|
|
On success, produces the unique name with that prefix. Otherwise,
|
|
throws a HTTPException.
|
|
"""
|
|
|
|
finder: PrefixFinder
|
|
|
|
@property
|
|
def responses(self) -> dict:
|
|
return {
|
|
**_RESPONSE_OK,
|
|
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
|