Coverage for dlabel/traefik_conf.py: 99%

1from dictknife import deepmerge 

2from pydantic import BaseModel, model_validator, BeforeValidator, ConfigDict 

3from typing import Any, Annotated 

4from logging import getLogger 

5 

6_log = getLogger(__name__) 

7 

8excludes = dict(exclude_none=True, exclude_defaults=True, exclude_unset=True) 

9 

10 

11class Model(BaseModel): 

12 """traefik configuration data model""" 

13 

14 model_config = ConfigDict(extra="allow") 

15 

16 @model_validator(mode="before") 

17 def __lowercase_property_keys__(cls, values: Any) -> Any: 

18 def __lower__(value: Any) -> Any: 

19 if isinstance(value, dict): 

20 return {k.lower(): __lower__(v) for k, v in value.items()} 

21 return value 

22 

23 return __lower__(values) 

24 

25 def merge(self, other: BaseModel): 

26 if other is None: 

27 return self 

28 _log.debug("merge: %s + %s", self, other) 

29 obj = deepmerge( 

30 self.model_dump(**excludes), other.model_dump(**excludes)) 

31 _log.debug("merged: %s", obj) 

32 return self.model_validate(obj) 

33 

34 def setbyaddr(self, address: list[str], value): 

35 _log.debug("set(addr) %s -> %s", address, value) 

36 res: dict[str, Any] = {} 

37 tgt = res 

38 for k in address[:-1]: 

39 tgt[k] = {} 

40 tgt = tgt[k] 

41 if value == "true": 

42 value = {} 

43 tgt[address[-1]] = value 

44 _log.debug("set-dict: %s", res) 

45 return self.merge(self.model_validate(res)) 

46 

47 def __str__(self) -> str: 

48 return self.model_dump_json(**excludes) 

49 

50 def to_dict(self) -> dict: 

51 return self.model_dump(exclude_unset=True, exclude_defaults=True, exclude_none=True) 

52 

53 

54def csv_list(v: Any) -> list[str]: 

55 if isinstance(v, str): 

56 return v.split(",") 

57 return v 

58 

59 

60ListStr = Annotated[list[str], BeforeValidator(csv_list)] 

61 

62 

63class CertFile(Model): 

64 certfile: str | None = None 

65 keyfile: str | None = None 

66 

67 

68class StoreCert(Model): 

69 defaultcertificate: CertFile | None = None 

70 defaultgeneratedcert: dict[str, Any] | None = None 

71 

72 

73class TlsCert(CertFile): 

74 stores: list[str] | None = None 

75 

76 

77class TlsConfig(Model): 

78 certificates: list[TlsCert] | None = None 

79 stores: dict[str, StoreCert] | None = None 

80 options: dict[str, Any] | None = None 

81 

82 

83class HttpRouter(Model): 

84 entrypoints: ListStr | None = None 

85 rule: str | None = None 

86 rulesyntax: str | None = None 

87 middlewares: ListStr | None = None 

88 service: str | None = None 

89 priority: int | None = None 

90 tls: dict[str, Any] | bool | None = None 

91 

92 

93class HttpLoadBalancerServer(Model): 

94 # incompatible with traefik configuration (used by docker label) 

95 host: str | None = None 

96 ipaddress: str | None = None 

97 port: int | None = None 

98 

99 

100class HttpLoadBalancerServers(Model): 

101 url: str | None = None 

102 

103 

104class HttpLoadBalancer(Model): 

105 servers: list[HttpLoadBalancerServers] | None = None 

106 server: HttpLoadBalancerServer | None = None 

107 sticky: dict[str, Any] | None = None 

108 healthcheck: dict[str, str] | None = None 

109 passhostheader: bool | None = None 

110 serverstransport: str | None = None 

111 responseforwarding: dict | None = None 

112 

113 

114class HttpService(Model): 

115 loadbalancer: HttpLoadBalancer | None = None 

116 weighted: dict[str, Any] | None = None 

117 mirroring: dict[str, Any] | None = None 

118 failover: dict[str, Any] | None = None 

119 

120 

121class CompressMiddleware(Model): 

122 excludedcontenttypes: list[str] | None = None 

123 includedcontenttypes: list[str] | None = None 

124 minresponsebodybytes: int | None = None 

125 defaultencoding: str | None = None 

126 encodings: ListStr | None = None 

127 

128 

129class HeadersMiddleware(Model): 

130 customrequestheaders: dict[str, str] | None = None 

131 customresponseheaders: dict[str, str] | None = None 

132 

133 

134class StripprefixMiddleware(Model): 

135 prefixes: ListStr | None = None 

136 forceslash: bool | None = None 

137 

138 

139class StripprefixregexMiddleware(Model): 

140 regex: ListStr | None = None 

141 

142 

143class AddprefixMiddleware(Model): 

144 prefix: str | None = None 

145 

146 

147class HttpMiddleware(Model): 

148 addprefix: AddprefixMiddleware | None = None 

149 basicauth: dict[str, Any] | None = None 

150 buffering: dict[str, Any] | None = None 

151 chain: dict[str, Any] | None = None 

152 circuitbreaker: dict[str, Any] | None = None 

153 compress: CompressMiddleware | bool | None = None 

154 contenttype: dict[str, Any] | bool | None = None 

155 digestauth: dict[str, Any] | None = None 

156 errors: dict[str, Any] | None = None 

157 forwardauth: dict[str, Any] | None = None 

158 grpcweb: dict[str, Any] | None = None 

159 headers: HeadersMiddleware | None = None 

160 ipwhitelist: dict[str, Any] | None = None 

161 ipallowlist: dict[str, Any] | None = None 

162 inflightreq: dict[str, Any] | None = None 

163 passtlsclientcert: dict[str, Any] | None = None 

164 ratelimit: dict[str, Any] | None = None 

165 redirectregex: dict[str, Any] | None = None 

166 redirectscheme: dict[str, Any] | None = None 

167 replacepath: dict[str, Any] | None = None 

168 replacepathregex: dict[str, Any] | None = None 

169 retry: dict[str, Any] | None = None 

170 stripprefix: StripprefixMiddleware | None = None 

171 stripprefixregex: StripprefixregexMiddleware | None = None 

172 

173 

174class HttpConfig(Model): 

175 middlewares: dict[str, HttpMiddleware] | None = None 

176 routers: dict[str, HttpRouter] | None = None 

177 services: dict[str, HttpService] | None = None 

178 serverstransports: dict[str, dict] | None = None 

179 

180 

181class AccessLogField(Model): 

182 defaultmode: str | None = None 

183 names: dict[str, str] | None = None 

184 headers: dict[str, Any] | None = None 

185 

186 

187class AccessLogConfig(Model): 

188 addinternals: bool = False 

189 filepath: str | None = None 

190 format: str | None = None 

191 bufferingsize: int | None = None 

192 filters: dict[str, Any] | None = None 

193 fields: AccessLogField | None = None 

194 

195 

196class EntrypointHttp(Model): 

197 redirections: dict[str, Any] | None = None 

198 encodequerysemicolons: bool | None = None 

199 middlewares: list[str] | None = None 

200 tls: dict[str, Any] | None = None 

201 

202 

203class EntrypointHttp2(Model): 

204 maxconcurrentstreams: int | None = None 

205 

206 

207class EntrypointHttp3(Model): 

208 advertisedport: int | None = None 

209 

210 

211class EntrypointConfig(Model): 

212 address: str | None = None 

213 http: EntrypointHttp | None = None 

214 http2: EntrypointHttp2 | None = None 

215 http3: EntrypointHttp3 | bool | None = None 

216 udp: dict[str, Any] | None = None 

217 allowacmebypass: bool | None = None 

218 reuseport: bool | None = None 

219 asdefault: bool | None = None 

220 forwardedheaders: dict[str, Any] | None = None 

221 transport: dict[str, Any] | None = None 

222 proxyprotocol: dict[str, Any] | None = None 

223 

224 

225class FileProvider(Model): 

226 filename: str | None = None 

227 directory: str | None = None 

228 watch: bool | None = None 

229 

230 

231class ProviderConfig(Model): 

232 docker: dict[str, Any] | bool | None = None 

233 file: FileProvider | None = None 

234 swarm: dict[str, Any] | None = None 

235 kubernetescrd: dict[str, Any] | None = None 

236 kubernetesingress: dict[str, Any] | bool | None = None 

237 kubernetesgateway: dict[str, Any] | bool | None = None 

238 consulcatalog: dict[str, Any] | bool | None = None 

239 nomad: dict[str, Any] | bool | None = None 

240 ecs: dict[str, Any] | bool | None = None 

241 consul: dict[str, Any] | None = None 

242 etcd: dict[str, Any] | None = None 

243 zookeeper: dict[str, Any] | None = None 

244 redis: dict[str, Any] | None = None 

245 http: dict[str, Any] | None = None 

246 

247 

248class TraefikConfig(Model): 

249 tls: TlsConfig | None = None 

250 http: HttpConfig | None = None 

251 tcp: dict[str, Any] | None = None 

252 udp: dict[str, Any] | None = None 

253 entrypoints: dict[str, EntrypointConfig] | None = None 

254 providers: ProviderConfig | None = None 

255 api: dict[str, Any] | None = None 

256 accesslog: dict[str, Any] | None = None 

257 experimental: dict[str, Any] | None = None 

258 log: dict[str, Any] | None = None 

259 metrics: dict[str, Any] | None = None 

260 tracing: dict[str, Any] | None = None 

261 certificatesresolvers: dict[str, Any] | None = None 

262 spiffe: dict[str, Any] | None = None