1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
|
"""JSONSchema spec handlers file module."""
from json import dumps
from json import loads
from typing import Any
from typing import ContextManager
from typing import Optional
from typing import Tuple
from urllib.parse import urlparse
from yaml import load
from jsonschema_path.handlers.protocols import SupportsRead
from jsonschema_path.handlers.utils import uri_to_path
from jsonschema_path.loaders import JsonschemaSafeLoader
class FileHandler:
"""File-like object handler."""
def __init__(self, loader: Any = JsonschemaSafeLoader):
self.loader = loader
def __call__(self, stream: SupportsRead) -> Any:
data = self._load(stream)
return loads(dumps(data))
def _load(self, stream: SupportsRead) -> Any:
return load(stream, self.loader)
class BaseFilePathHandler:
"""Base file path handler."""
allowed_schemes: Tuple[str, ...] = NotImplemented
def __init__(
self, *allowed_schemes: str, file_handler: Optional[FileHandler] = None
):
self.allowed_schemes = allowed_schemes or self.allowed_schemes
self.file_handler = file_handler or FileHandler()
def __call__(self, uri: str) -> Any:
parsed_url = urlparse(uri)
if parsed_url.scheme not in self.allowed_schemes:
raise ValueError(f"Scheme {parsed_url.scheme} not allowed")
with self._open(uri) as stream:
return self.file_handler(stream)
def _open(self, uri: str) -> ContextManager[SupportsRead]:
raise NotImplementedError
class FilePathHandler(BaseFilePathHandler):
"""File path handler."""
allowed_schemes = ("file",)
def __init__(
self,
*allowed_schemes: str,
file_handler: Optional[FileHandler] = None,
encoding: str = "utf-8",
):
super().__init__(*allowed_schemes, file_handler=file_handler)
self.encoding = encoding
def _open(self, uri: str) -> ContextManager[SupportsRead]:
filepath = uri_to_path(uri)
return open(filepath, encoding=self.encoding)
|
