File size: 2,795 Bytes
08c964e | 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 72 73 74 75 76 77 78 79 | from dataclasses import dataclass, field
from typing import List, Optional, Union
from .nginx_base import Block, Directive, _Trans, IDirective, trans_
from .nginx_components import Server, Http, Upstream
@dataclass
class Config(Block):
"""配置"""
file_path: str = ""
def find_servers(self) -> List['Server']:
"""查找所有server块"""
servers = []
directives = self.find_directives("server", include=True)
for directive in directives:
# 使用字符串类型检查避免循环依赖
if directive.__class__.__name__ == 'Server':
servers.append(directive)
return servers
def find_http(self) -> Optional['Http']:
"""查找http块"""
directives = self.find_directives("http")
if directives:
directive = directives[0]
# 使用字符串类型检查避免循环依赖
if directive.__class__.__name__ == 'Http':
return trans_(directive, Http)
return None
def find_upstreams(self) -> List['Upstream']:
"""查找所有upstream块"""
upstreams = []
directives = self.find_directives("upstream")
for directive in directives:
# 使用字符串类型检查避免循环依赖
if directive.__class__.__name__ == 'Upstream':
upstreams.append(directive)
return upstreams
@dataclass
class Include(Directive, _Trans): # 不实现 IBlock 接口，但可以解析
"""包含文件"""
include_path: str = ""
configs: List[Config] = field(default_factory=list)
@classmethod
def from_directive(cls, directive: Union[IDirective, Directive]) -> 'Include':
if directive.__class__ is cls:
return trans_(directive, cls)
parameters = directive.get_parameters()
if not len(parameters) == 1:
raise ValueError("include指令参数数量错误")
return cls(
line=directive.get_line(),
block=directive.get_block(),
name=directive.get_name(),
comment=directive.get_comment(),
inline_comment=directive.get_inline_comment(),
parameters=parameters,
_parent=directive.get_parent(),
include_path=parameters[0]
)
def get_directives(self) -> List[IDirective]:
res =[]
for config in self.configs:
res.extend(config.get_directives())
return res
def find_directives(self, directive_name: str, include: bool = False, sub_block: bool = False) -> List[IDirective]:
res =[]
for config in self.configs:
res.extend(config.find_directives(directive_name, include=include, sub_block=sub_block))
return res
|