Coverage for dlabel/traefik_conf.py: 99%

1from logging import getLogger 

2from typing import Annotated, Any 

3 

4from dictknife import deepmerge 

5from pydantic import BaseModel, BeforeValidator, ConfigDict, model_validator 

6 

7_log = getLogger(__name__) 

8 

9excludes = {"exclude_none": True, "exclude_defaults": True, "exclude_unset": True} 

10 

11 

12class Model(BaseModel): 

13 """traefik configuration data model""" 

14 

15 model_config = ConfigDict(extra="allow") 

16 

17 @model_validator(mode="before") 

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

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

20 if isinstance(value, dict): 

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

22 return value 

23 

24 return __lower__(values) 

25 

26 def merge(self, other: BaseModel): 

27 if other is None: 

28 return self 

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

30 obj = deepmerge(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( 

52 exclude_unset=True, exclude_defaults=True, exclude_none=True 

53 ) 

54 

55 

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

57 if isinstance(v, str): 

58 return v.split(",") 

59 return v 

60 

61 

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

63 

64 

65class CertFile(Model): 

66 certfile: str | None = None 

67 keyfile: str | None = None 

68 

69 

70class StoreCert(Model): 

71 defaultcertificate: CertFile | None = None 

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

73 

74 

75class TlsCert(CertFile): 

76 stores: list[str] | None = None 

77 

78 

79class TlsConfig(Model): 

80 certificates: list[TlsCert] | None = None 

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

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

83 

84 

85class HttpRouter(Model): 

86 entrypoints: ListStr | None = None 

87 rule: str | None = None 

88 rulesyntax: str | None = None 

89 middlewares: ListStr | None = None 

90 service: str | None = None 

91 priority: int | None = None 

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

93 

94 

95class HttpLoadBalancerServer(Model): 

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

97 host: str | None = None 

98 ipaddress: str | None = None 

99 port: int | None = None 

100 

101 

102class HttpLoadBalancerServers(Model): 

103 url: str | None = None 

104 

105 

106class HttpLoadBalancer(Model): 

107 servers: list[HttpLoadBalancerServers] | None = None 

108 server: HttpLoadBalancerServer | None = None 

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

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

111 passhostheader: bool | None = None 

112 serverstransport: str | None = None 

113 responseforwarding: dict | None = None 

114 

115 

116class HttpService(Model): 

117 loadbalancer: HttpLoadBalancer | None = None 

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

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

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

121 

122 

123class CompressMiddleware(Model): 

124 excludedcontenttypes: list[str] | None = None 

125 includedcontenttypes: list[str] | None = None 

126 minresponsebodybytes: int | None = None 

127 defaultencoding: str | None = None 

128 encodings: ListStr | None = None 

129 

130 

131class HeadersMiddleware(Model): 

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

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

134 

135 

136class StripprefixMiddleware(Model): 

137 prefixes: ListStr | None = None 

138 forceslash: bool | None = None 

139 

140 

141class StripprefixregexMiddleware(Model): 

142 regex: ListStr | None = None 

143 

144 

145class AddprefixMiddleware(Model): 

146 prefix: str | None = None 

147 

148 

149class HttpMiddleware(Model): 

150 addprefix: AddprefixMiddleware | None = None 

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

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

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

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

155 compress: CompressMiddleware | bool | None = None 

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

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

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

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

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

161 headers: HeadersMiddleware | None = None 

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

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

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

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

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

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

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

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

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

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

172 stripprefix: StripprefixMiddleware | None = None 

173 stripprefixregex: StripprefixregexMiddleware | None = None 

174 

175 

176class HttpConfig(Model): 

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

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

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

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

181 

182 

183class AccessLogField(Model): 

184 defaultmode: str | None = None 

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

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

187 

188 

189class AccessLogConfig(Model): 

190 addinternals: bool = False 

191 filepath: str | None = None 

192 format: str | None = None 

193 bufferingsize: int | None = None 

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

195 fields: AccessLogField | None = None 

196 

197 

198class EntrypointHttp(Model): 

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

200 encodequerysemicolons: bool | None = None 

201 middlewares: list[str] | None = None 

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

203 

204 

205class EntrypointHttp2(Model): 

206 maxconcurrentstreams: int | None = None 

207 

208 

209class EntrypointHttp3(Model): 

210 advertisedport: int | None = None 

211 

212 

213class EntrypointConfig(Model): 

214 address: str | None = None 

215 http: EntrypointHttp | None = None 

216 http2: EntrypointHttp2 | None = None 

217 http3: EntrypointHttp3 | bool | None = None 

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

219 allowacmebypass: bool | None = None 

220 reuseport: bool | None = None 

221 asdefault: bool | None = None 

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

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

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

225 

226 

227class FileProvider(Model): 

228 filename: str | None = None 

229 directory: str | None = None 

230 watch: bool | None = None 

231 

232 

233class ProviderConfig(Model): 

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

235 file: FileProvider | None = None 

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

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

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

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

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

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

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

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

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

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

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

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

248 

249 

250class TraefikConfig(Model): 

251 tls: TlsConfig | None = None 

252 http: HttpConfig | None = None 

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

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

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

256 providers: ProviderConfig | None = None 

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

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

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

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

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

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

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

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