from knot_resolver_manager.datamodel.types import Dir, DomainName, File, IntNonNegative, Percent, SizeUnit, TimeUnit
from knot_resolver_manager.utils.modeling import ConfigSchema
+from knot_resolver_manager.utils.modeling.base_schema import lazy_default
class PrefillSchema(ConfigSchema):
prefill: Prefill the cache periodically by importing zone data obtained over HTTP.
"""
- storage: Dir = Dir("/var/cache/knot-resolver")
+ storage: Dir = lazy_default(Dir, "/var/cache/knot-resolver")
size_max: SizeUnit = SizeUnit("100M")
garbage_collector: Union[GarbageCollectorSchema, Literal[False]] = GarbageCollectorSchema()
ttl_min: TimeUnit = TimeUnit("5s")
from knot_resolver_manager.datamodel.options_schema import OptionsSchema
from knot_resolver_manager.datamodel.policy_schema import PolicySchema
from knot_resolver_manager.datamodel.slice_schema import SliceSchema
-from knot_resolver_manager.datamodel.types import IntPositive
-from knot_resolver_manager.datamodel.types.files import UncheckedPath
+from knot_resolver_manager.datamodel.types import Dir, IntPositive
from knot_resolver_manager.datamodel.view_schema import ViewSchema
from knot_resolver_manager.datamodel.webmgmt_schema import WebmgmtSchema
from knot_resolver_manager.utils.modeling import ConfigSchema
version: int = 1
nsid: Optional[str] = None
hostname: Optional[str] = None
- rundir: UncheckedPath = UncheckedPath("/var/run/knot-resolver")
+ rundir: Dir = lazy_default(Dir, "/var/run/knot-resolver")
workers: Union[Literal["auto"], IntPositive] = IntPositive(1)
max_workers: IntPositive = IntPositive(_default_max_worker_count())
management: ManagementSchema = lazy_default(ManagementSchema, {"unix-socket": "./manager.sock"})
slices: Optional[List[SliceSchema]] = None
policy: Optional[List[PolicySchema]] = None
forward: Optional[List[ForwardSchema]] = None
- cache: CacheSchema = CacheSchema()
+ cache: CacheSchema = lazy_default(CacheSchema, {})
dnssec: Union[bool, DnssecSchema] = True
dns64: Union[bool, Dns64Schema] = False
logging: LoggingSchema = LoggingSchema()
nsid: Optional[str]
hostname: str
- rundir: UncheckedPath
+ rundir: Dir
workers: IntPositive
max_workers: IntPositive
management: ManagementSchema
return _MAIN_TEMPLATE.render(cfg=self, cwd=os.getcwd()) # pyright: reportUnknownMemberType=false
-def get_rundir_without_validation(data: Dict[str, Any]) -> UncheckedPath:
+def get_rundir_without_validation(data: Dict[str, Any]) -> Dir:
"""
Without fully parsing, try to get a rundir from a raw config data. When it fails,
attempts a full validation to produce a good error message.
_ = KresConfig(data) # this should throw a descriptive error
assert False
- return UncheckedPath(rundir, object_path="/rundir")
+ return Dir(rundir, object_path="/rundir")
from typing_extensions import Literal
-from knot_resolver_manager.datamodel.types import DomainName, IPAddressOptionalPort, ListOrItem
-from knot_resolver_manager.datamodel.types.files import FilePath
+from knot_resolver_manager.datamodel.types import DomainName, File, IPAddressOptionalPort, ListOrItem
from knot_resolver_manager.utils.modeling import ConfigSchema
transport: Optional[Literal["tls"]] = None
pin_sha256: Optional[ListOrItem[str]] = None
hostname: Optional[DomainName] = None
- ca_file: Optional[FilePath] = None
+ ca_file: Optional[File] = None
def _validate(self) -> None:
if self.pin_sha256 and (self.hostname or self.ca_file):
from typing_extensions import Literal
-from knot_resolver_manager.datamodel.types import DomainName, IDPattern, IPAddress, ListOrItem, TimeUnit
-from knot_resolver_manager.datamodel.types.files import FilePath
+from knot_resolver_manager.datamodel.types import DomainName, File, IDPattern, IPAddress, ListOrItem, TimeUnit
from knot_resolver_manager.utils.modeling import ConfigSchema
nodata: bool = True
addresses: Optional[List[IPAddress]] = None
roots: Optional[List[DomainName]] = None
- roots_file: Optional[FilePath] = None
+ roots_file: Optional[File] = None
roots_url: Optional[str] = None
refresh: Optional[TimeUnit] = None
tags: Tags to link with other policy rules.
"""
- file: FilePath
+ file: File
tags: Optional[List[IDPattern]] = None
ttl: Optional[TimeUnit] = None
nodata: bool = True
root_fallback_addresses: Optional[Dict[DomainName, ListOrItem[IPAddress]]] = None
- root_fallback_addresses_files: Optional[ListOrItem[FilePath]] = None
+ root_fallback_addresses_files: Optional[ListOrItem[File]] = None
addresses: Optional[Dict[DomainName, IPAddress]] = None
- addresses_files: Optional[ListOrItem[FilePath]] = None
+ addresses_files: Optional[ListOrItem[File]] = None
records: Optional[str] = None
subtrees: Optional[List[SubtreeSchema]] = None
rpz: Optional[List[RPZSchema]] = None
def _set_working_directory(config_raw: Dict[str, Any]) -> None:
rundir = get_rundir_without_validation(config_raw)
- if not rundir.to_path().exists():
- raise KresManagerException(f"`rundir` directory ({rundir}) does not exist!")
-
logger.info("changing working directory to rundir at '%s'", rundir.to_path().absolute())
os.chdir(rundir.to_path())
# before processing any configuration, set validation context
# - resolve_root = root against which all relative paths will be resolved
- set_global_validation_context(Context(resolve_directory=config.parent))
+ set_global_validation_context(Context(config.parent))
# Preprocess config - load from file or in general take it to the last step before validation.
config_raw = await _load_raw_config(config)
from knot_resolver_manager.datamodel.globals import Context, set_global_validation_context
-set_global_validation_context(Context(Path(".")))
+set_global_validation_context(Context(Path("."), False))