confighelper: avoid direct import of JinjaTemplate

Signed-off-by:  Eric Callahan <arksine.code@gmail.com>
This commit is contained in:
Eric Callahan 2024-01-13 12:37:35 -05:00
parent ddd735feba
commit 60f4a82873
No known key found for this signature in database
GPG Key ID: 5A1EB336DFB4C71B
4 changed files with 26 additions and 13 deletions

View File

@ -11,6 +11,7 @@ import copy
import re
from enum import Enum, Flag, auto
from dataclasses import dataclass
from abc import ABCMeta, abstractmethod
from .utils import ServerError, Sentinel
from .utils import json_wrapper as jsonw
@ -159,6 +160,19 @@ class KlippyState(ExtendedEnum):
def startup_complete(self) -> bool:
return self.value > 2
class RenderableTemplate(metaclass=ABCMeta):
@abstractmethod
def __str__(self) -> str:
...
@abstractmethod
def render(self, context: Dict[str, Any] = {}) -> str:
...
@abstractmethod
async def render_async(self, context: Dict[str, Any] = {}) -> str:
...
@dataclass(frozen=True)
class APIDefinition:
endpoint: str

View File

@ -33,7 +33,6 @@ if TYPE_CHECKING:
from .machine import Machine
from .klippy_apis import KlippyAPI as APIComp
from .mqtt import MQTTClient
from .template import JinjaTemplate
from .http_client import HttpClient
from .klippy_connection import KlippyConnection
@ -1207,7 +1206,7 @@ class MQTTDevice(PowerDevice):
self.mqtt: MQTTClient = self.server.load_component(config, 'mqtt')
self.eventloop = self.server.get_event_loop()
self.cmd_topic: str = config.get('command_topic')
self.cmd_payload: JinjaTemplate = config.gettemplate('command_payload')
self.cmd_payload = config.gettemplate('command_payload')
self.retain_cmd_state = config.getboolean('retain_command_state', False)
self.query_topic: Optional[str] = config.get('query_topic', None)
self.query_payload = config.gettemplate('query_payload', None)

View File

@ -8,6 +8,7 @@ import logging
import asyncio
import jinja2
from ..utils import json_wrapper as jsonw
from ..common import RenderableTemplate
# Annotation imports
from typing import (
@ -70,7 +71,7 @@ class TemplateFactory:
return JinjaTemplate(source, self.server, template, True)
class JinjaTemplate:
class JinjaTemplate(RenderableTemplate):
def __init__(self,
source: str,
server: Server,

View File

@ -15,7 +15,7 @@ import copy
import logging
from io import StringIO
from .utils import Sentinel
from .components.template import JinjaTemplate
from .common import RenderableTemplate
# Annotation imports
from typing import (
@ -408,19 +408,18 @@ class ConfigHelper:
default: Union[Sentinel, _T] = Sentinel.MISSING,
is_async: bool = False,
deprecate: bool = False
) -> Union[JinjaTemplate, _T]:
) -> Union[RenderableTemplate, _T]:
try:
template: TemplateFactory
template = self.server.load_component(self, 'template')
template: TemplateFactory = self.server.load_component(self, 'template')
except Exception:
raise ConfigError(
f"Section [{self.section}], option '{option}', "
"Template Component not available")
f"Section [{self.section}], option '{option}': "
"Failed to load 'template' component."
)
def gettemplate_wrapper(sec: str, opt: str) -> JinjaTemplate:
def gettemplate_wrapper(sec: str, opt: str) -> RenderableTemplate:
val = self.config.get(sec, opt)
return template.create_template(val.strip(), is_async)
return self._get_option(gettemplate_wrapper, option, default,
deprecate=deprecate)
@ -429,7 +428,7 @@ class ConfigHelper:
default: Union[Sentinel, str] = Sentinel.MISSING,
is_async: bool = False,
deprecate: bool = False
) -> JinjaTemplate:
) -> RenderableTemplate:
val = self.gettemplate(option, default, is_async, deprecate)
if isinstance(val, str):
template: TemplateFactory
@ -443,7 +442,7 @@ class ConfigHelper:
deprecate: bool = False
) -> Union[pathlib.Path, _T]:
val = self.gettemplate(option, default, deprecate=deprecate)
if isinstance(val, JinjaTemplate):
if isinstance(val, RenderableTemplate):
ctx = {"data_path": self.server.get_app_args()["data_path"]}
strpath = val.render(ctx)
return pathlib.Path(strpath).expanduser().resolve()