402 lines
13 KiB
Python
402 lines
13 KiB
Python
#!/usr/bin/env python3.9
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
JFinal Engine - Template Engine Main Class
|
|
"""
|
|
|
|
from typing import Dict, Optional, Set, Type, Any, Callable
|
|
from .EngineConfig import EngineConfig
|
|
from .Env import Env
|
|
from .Template import Template
|
|
from .stat.Parser import Parser
|
|
from .stat.ast.Stat import Stat
|
|
from .source.ISourceFactory import ISourceFactory
|
|
from .source.ClassPathSourceFactory import ClassPathSourceFactory
|
|
from .source.StringSource import StringSource
|
|
from .io.WriterBuffer import WriterBuffer
|
|
from .io.EncoderFactory import EncoderFactory
|
|
from .io.JdkEncoderFactory import JdkEncoderFactory
|
|
from .stat.Compressor import Compressor
|
|
from .expr.ast.FieldKit import FieldKit
|
|
from .expr.ast.FieldKeyBuilder import FieldKeyBuilder
|
|
from .expr.ast.MethodKit import MethodKit
|
|
from .Directive import Directive
|
|
from .stat.OutputDirectiveFactory import OutputDirectiveFactory
|
|
from ..kit.SyncWriteMap import SyncWriteMap
|
|
from ..kit.StrKit import StrKit
|
|
from ..kit.HashKit import HashKit
|
|
|
|
class Engine:
|
|
"""Template Engine - Main class for JFinal Template Engine"""
|
|
|
|
MAIN_ENGINE_NAME = "main"
|
|
|
|
_MAIN_ENGINE: 'Engine' = None
|
|
_engine_map: Dict[str, 'Engine'] = {}
|
|
|
|
def __init__(self, engine_name: str = None):
|
|
"""
|
|
Initialize engine
|
|
|
|
Args:
|
|
engine_name: Engine name (optional)
|
|
"""
|
|
self._name = engine_name or "NO_NAME"
|
|
self._dev_mode = True
|
|
self._cache_string_template = False
|
|
self._config = EngineConfig()
|
|
self._source_factory = self._config.source_factory
|
|
self._template_cache = SyncWriteMap(2048, 0.5)
|
|
|
|
@classmethod
|
|
def use(cls, engine_name: str = None) -> 'Engine':
|
|
"""
|
|
Get engine instance
|
|
|
|
Args:
|
|
engine_name: Engine name (optional, uses main engine if not specified)
|
|
|
|
Returns:
|
|
Engine instance
|
|
"""
|
|
if engine_name is None:
|
|
if cls._MAIN_ENGINE is None:
|
|
cls._MAIN_ENGINE = cls(cls.MAIN_ENGINE_NAME)
|
|
cls._engine_map[cls.MAIN_ENGINE_NAME] = cls._MAIN_ENGINE
|
|
return cls._MAIN_ENGINE
|
|
|
|
return cls._engine_map.get(engine_name)
|
|
|
|
@classmethod
|
|
def create(cls, engine_name: str, configurator: Callable[['Engine'], None] = None) -> 'Engine':
|
|
"""
|
|
Create new engine with name
|
|
|
|
Args:
|
|
engine_name: Engine name
|
|
configurator: Configuration callback
|
|
|
|
Returns:
|
|
New engine instance
|
|
"""
|
|
if StrKit.is_blank(engine_name):
|
|
raise ValueError("Engine name cannot be blank")
|
|
|
|
engine_name = engine_name.strip()
|
|
if engine_name in cls._engine_map:
|
|
raise ValueError(f"Engine already exists: {engine_name}")
|
|
|
|
new_engine = cls(engine_name)
|
|
cls._engine_map[engine_name] = new_engine
|
|
|
|
if configurator:
|
|
configurator(new_engine)
|
|
|
|
return new_engine
|
|
|
|
@classmethod
|
|
def create_if_absent(cls, engine_name: str, configurator: Callable[['Engine'], None] = None) -> 'Engine':
|
|
"""
|
|
Create engine if absent
|
|
|
|
Args:
|
|
engine_name: Engine name
|
|
configurator: Configuration callback
|
|
|
|
Returns:
|
|
Engine instance
|
|
"""
|
|
ret = cls._engine_map.get(engine_name)
|
|
if ret is None:
|
|
# Double-check locking
|
|
if engine_name not in cls._engine_map:
|
|
ret = cls.create(engine_name)
|
|
if configurator:
|
|
configurator(ret)
|
|
return ret
|
|
|
|
@classmethod
|
|
def remove(cls, engine_name: str) -> Optional['Engine']:
|
|
"""
|
|
Remove engine by name
|
|
|
|
Args:
|
|
engine_name: Engine name
|
|
|
|
Returns:
|
|
Removed engine or None
|
|
"""
|
|
removed = cls._engine_map.pop(engine_name, None)
|
|
if removed and cls.MAIN_ENGINE_NAME == removed.name:
|
|
cls._MAIN_ENGINE = None
|
|
return removed
|
|
|
|
@classmethod
|
|
def set_main_engine(cls, engine: 'Engine'):
|
|
"""
|
|
Set main engine
|
|
|
|
Args:
|
|
engine: Engine to set as main
|
|
"""
|
|
if engine is None:
|
|
raise ValueError("Engine cannot be null")
|
|
|
|
engine._name = cls.MAIN_ENGINE_NAME
|
|
cls._engine_map[cls.MAIN_ENGINE_NAME] = engine
|
|
cls._MAIN_ENGINE = engine
|
|
|
|
def get_template(self, file_name: str) -> Template:
|
|
"""
|
|
Get template by file name
|
|
|
|
Args:
|
|
file_name: Template file name
|
|
|
|
Returns:
|
|
Template instance
|
|
"""
|
|
template = self._template_cache.get(file_name)
|
|
|
|
if template is None:
|
|
template = self._build_template_by_source_factory(file_name)
|
|
self._template_cache[file_name] = template
|
|
elif self._dev_mode and template.is_modified():
|
|
template = self._build_template_by_source_factory(file_name)
|
|
self._template_cache[file_name] = template
|
|
|
|
return template
|
|
|
|
def _build_template_by_source_factory(self, file_name: str) -> Template:
|
|
"""Build template using source factory"""
|
|
source = self._source_factory.get_source(
|
|
self._config.base_template_path,
|
|
file_name,
|
|
self._config.encoding
|
|
)
|
|
|
|
return self._build_template(source, file_name)
|
|
|
|
def get_template_by_string(self, content: str, cache: bool = False) -> Template:
|
|
"""
|
|
Get template by string content
|
|
|
|
Args:
|
|
content: Template content
|
|
cache: Whether to cache the template
|
|
|
|
Returns:
|
|
Template instance
|
|
"""
|
|
if not cache:
|
|
source = StringSource(content, False)
|
|
return self._build_template(source, None)
|
|
|
|
cache_key = HashKit.md5(content)
|
|
template = self._template_cache.get(cache_key)
|
|
|
|
if template is None:
|
|
source = StringSource(content, cache_key)
|
|
template = self._build_template(source, cache_key)
|
|
self._template_cache[cache_key] = template
|
|
elif self._dev_mode and template.is_modified():
|
|
source = StringSource(content, cache_key)
|
|
template = self._build_template(source, cache_key)
|
|
self._template_cache[cache_key] = template
|
|
|
|
return template
|
|
|
|
def _build_template(self, source, cache_key: str = None) -> Template:
|
|
"""Build template from source"""
|
|
env = Env(self._config)
|
|
|
|
if self._dev_mode:
|
|
env.add_source(source)
|
|
|
|
parser = Parser(env, source.get_content(), cache_key)
|
|
stat = parser.parse()
|
|
|
|
return Template(env, stat)
|
|
|
|
def add_shared_function(self, file_name: str) -> 'Engine':
|
|
"""Add shared function from file"""
|
|
self._config.add_shared_function(file_name)
|
|
return self
|
|
|
|
def add_shared_function_by_string(self, content: str) -> 'Engine':
|
|
"""Add shared function from string"""
|
|
self._config.add_shared_function_by_content(content)
|
|
return self
|
|
|
|
def add_shared_object(self, name: str, obj: Any) -> 'Engine':
|
|
"""Add shared object"""
|
|
self._config.add_shared_object(name, obj)
|
|
return self
|
|
|
|
def remove_shared_object(self, name: str) -> 'Engine':
|
|
"""Remove shared object"""
|
|
self._config.remove_shared_object(name)
|
|
return self
|
|
|
|
def add_directive(self, directive_name: str, directive_class: Type[Directive],
|
|
keep_line_blank: bool = False) -> 'Engine':
|
|
"""Add custom directive"""
|
|
self._config.add_directive(directive_name, directive_class, keep_line_blank)
|
|
return self
|
|
|
|
def remove_directive(self, directive_name: str) -> 'Engine':
|
|
"""Remove directive"""
|
|
self._config.remove_directive(directive_name)
|
|
return self
|
|
|
|
def add_shared_method(self, obj: Any) -> 'Engine':
|
|
"""Add shared method from object"""
|
|
self._config.add_shared_method(obj)
|
|
return self
|
|
|
|
def add_shared_method_from_class(self, clazz: Type) -> 'Engine':
|
|
"""Add shared method from class"""
|
|
self._config.add_shared_method_from_class(clazz)
|
|
return self
|
|
|
|
def add_shared_static_method(self, clazz: Type) -> 'Engine':
|
|
"""Add shared static method from class"""
|
|
self._config.add_shared_static_method(clazz)
|
|
return self
|
|
|
|
def remove_shared_method(self, method_name: str) -> 'Engine':
|
|
"""Remove shared method by name"""
|
|
self._config.remove_shared_method(method_name)
|
|
return self
|
|
|
|
def remove_template_cache(self, cache_key: str):
|
|
"""Remove template cache by key"""
|
|
self._template_cache.pop(cache_key, None)
|
|
|
|
def remove_all_template_cache(self):
|
|
"""Remove all template cache"""
|
|
self._template_cache.clear()
|
|
|
|
@property
|
|
def dev_mode(self) -> bool:
|
|
"""Get dev mode setting"""
|
|
return self._dev_mode
|
|
|
|
@dev_mode.setter
|
|
def dev_mode(self, value: bool):
|
|
"""Set dev mode"""
|
|
self._dev_mode = value
|
|
self._config.dev_mode = value
|
|
if self._dev_mode:
|
|
self.remove_all_template_cache()
|
|
|
|
@property
|
|
def cache_string_template(self) -> bool:
|
|
"""Get cache string template setting"""
|
|
return self._cache_string_template
|
|
|
|
@cache_string_template.setter
|
|
def cache_string_template(self, value: bool):
|
|
"""Set cache string template"""
|
|
self._cache_string_template = value
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
"""Get engine name"""
|
|
return self._name
|
|
|
|
@property
|
|
def config(self) -> EngineConfig:
|
|
"""Get engine configuration"""
|
|
return self._config
|
|
|
|
@property
|
|
def source_factory(self) -> ISourceFactory:
|
|
"""Get source factory"""
|
|
return self._source_factory
|
|
|
|
@source_factory.setter
|
|
def source_factory(self, value: ISourceFactory):
|
|
"""Set source factory"""
|
|
self._config.source_factory = value
|
|
self._source_factory = value
|
|
|
|
def set_to_class_path_source_factory(self) -> 'Engine':
|
|
"""Set to class path source factory"""
|
|
self.source_factory = ClassPathSourceFactory()
|
|
return self
|
|
|
|
@property
|
|
def base_template_path(self) -> Optional[str]:
|
|
"""Get base template path"""
|
|
return self._config.base_template_path
|
|
|
|
@base_template_path.setter
|
|
def base_template_path(self, value: str):
|
|
"""Set base template path"""
|
|
self._config.base_template_path = value
|
|
|
|
@property
|
|
def encoding(self) -> str:
|
|
"""Get encoding"""
|
|
return self._config.encoding
|
|
|
|
@encoding.setter
|
|
def encoding(self, value: str):
|
|
"""Set encoding"""
|
|
self._config.encoding = value
|
|
|
|
@property
|
|
def date_pattern(self) -> str:
|
|
"""Get date pattern"""
|
|
return self._config.date_pattern
|
|
|
|
@date_pattern.setter
|
|
def date_pattern(self, value: str):
|
|
"""Set date pattern"""
|
|
self._config.date_pattern = value
|
|
|
|
def set_compressor(self, compressor: Compressor) -> 'Engine':
|
|
"""Set compressor"""
|
|
self._config._compressor = compressor
|
|
return self
|
|
|
|
def set_compressor_on(self, separator: str = '\n') -> 'Engine':
|
|
"""Set compressor on"""
|
|
self._config._compressor = Compressor(separator)
|
|
return self
|
|
|
|
def set_encoder_factory(self, encoder_factory: EncoderFactory) -> 'Engine':
|
|
"""Set encoder factory"""
|
|
self._config.writer_buffer.set_encoder_factory(encoder_factory)
|
|
self._config.writer_buffer.set_encoding(self._config.encoding)
|
|
return self
|
|
|
|
def set_to_jdk_encoder_factory(self) -> 'Engine':
|
|
"""Set to JDK encoder factory"""
|
|
self.set_encoder_factory(JdkEncoderFactory())
|
|
return self
|
|
|
|
def set_buffer_size(self, buffer_size: int) -> 'Engine':
|
|
"""Set buffer size"""
|
|
self._config.writer_buffer.set_buffer_size(buffer_size)
|
|
return self
|
|
|
|
def set_reentrant_buffer_size(self, reentrant_buffer_size: int) -> 'Engine':
|
|
"""Set reentrant buffer size"""
|
|
self._config.writer_buffer.set_reentrant_buffer_size(reentrant_buffer_size)
|
|
return self
|
|
|
|
@property
|
|
def template_cache_size(self) -> int:
|
|
"""Get template cache size"""
|
|
return len(self._template_cache)
|
|
|
|
def __repr__(self) -> str:
|
|
return f"Template Engine: {self._name}"
|
|
|
|
|
|
# Initialize main engine
|
|
Engine._MAIN_ENGINE = Engine(Engine.MAIN_ENGINE_NAME)
|
|
Engine._engine_map[Engine.MAIN_ENGINE_NAME] = Engine._MAIN_ENGINE
|