Coverage for src/dcm/models.py: 81%

1from enum import Enum 

2from typing import Any, Optional, TypeVar, Union 

3 

4from . import spec 

5 

6 

7def parse_list_of_dict_of_tuples( 

8 list_or_dict: Optional[ 

9 Union[dict[str, Any], list[Any], spec.ListOrDict, spec.ListOrDict1] 

10 ], 

11) -> dict[str, Any]: 

12 if isinstance(list_or_dict, dict): 

13 return list_or_dict 

14 if isinstance(list_or_dict, list): 

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

16 for e in list_or_dict: 

17 if "=" in str(e): 

18 x = str(e).split("=", maxsplit=1) 

19 res[x[0]] = x[1] 

20 return res 

21 if isinstance(list_or_dict, (spec.ListOrDict, spec.ListOrDict1)): 

22 return parse_list_of_dict_of_tuples(list_or_dict.root) 

23 assert list_or_dict is None 

24 return {} 

25 

26 

27class Protocol(str, Enum): 

28 # pylint: disable=invalid-name 

29 tcp = "tcp" 

30 udp = "udp" 

31 any = "any" 

32 

33 def __repr__(self) -> str: 

34 return self.name 

35 

36 

37class AppProtocol(str, Enum): 

38 # pylint: disable=invalid-name 

39 rest = "REST" 

40 mqtt = "MQTT" 

41 wbsock = "WebSocket" 

42 http = "http" 

43 https = "https" 

44 na = "NA" 

45 

46 def __repr__(self) -> str: 

47 return self.name 

48 

49 

50class Port: 

51 # pylint: disable=too-many-arguments,too-many-positional-arguments 

52 def __init__( 

53 self, 

54 source_file: str, 

55 host: str, 

56 source_port: str, 

57 container_port: str, 

58 protocol: Protocol = Protocol.any, 

59 app_protocol: AppProtocol = AppProtocol.na, 

60 ): 

61 self.host = host 

62 self.source_port = source_port 

63 self.container_port = container_port 

64 self.protocol = protocol 

65 self.app_protocol = app_protocol 

66 self.source_files: list[str] = [source_file] 

67 

68 

69class VolumeType(str, Enum): 

70 # pylint: disable=invalid-name 

71 volume = "volume" 

72 bind = "bind" 

73 tmpfs = "tmpfs" 

74 npipe = "npipe" 

75 

76 

77class Volume: 

78 # pylint: disable=too-many-arguments,too-many-positional-arguments 

79 def __init__( 

80 self, 

81 source_file: str, 

82 source: str, 

83 target: str, 

84 v_type: VolumeType = VolumeType.volume, 

85 access_mode: str = "rw", 

86 ): 

87 self.source = source 

88 self.target = target 

89 self.v_type = v_type 

90 self.access_mode = access_mode 

91 self.source_files: list[str] = [source_file] 

92 

93 

94class Device: 

95 def __init__( 

96 self, 

97 source_file: str, 

98 host_path: str, 

99 container_path: str, 

100 cgroup_permissions: Optional[str] = None, 

101 ): 

102 self.host_path = host_path 

103 self.container_path = container_path 

104 self.cgroup_permissions = cgroup_permissions 

105 self.source_files: list[str] = [source_file] 

106 

107 

108class Extends: 

109 def __init__(self, service_name: str, from_file: Optional[str] = None): 

110 self.service_name = service_name 

111 self.from_file = from_file 

112 

113 

114T = TypeVar("T") 

115 

116 

117def opt_to_arr(opt: Optional[list[T]]) -> list[T]: 

118 if opt is None: 

119 return [] 

120 return opt 

121 

122 

123def opt_to_dict(opt: Optional[dict[str, T]]) -> dict[str, T]: 

124 if opt is None: 

125 return {} 

126 return opt 

127 

128 

129def _parse_external(ext: Optional[Union[bool, spec.External]]) -> bool: 

130 if ext is None: 

131 return False 

132 if isinstance(ext, bool): 

133 return ext 

134 return str(ext).lower() == "true" 

135 

136 

137class EnvFileInfo: 

138 def __init__( 

139 self, 

140 path: str, 

141 required: Optional[bool] = True, 

142 ) -> None: 

143 self.path = path 

144 self.required = True if required is None else required 

145 

146 

147class Service: 

148 # pylint: disable=too-many-arguments,too-many-instance-attributes,too-many-positional-arguments,too-many-locals 

149 def __init__( 

150 self, 

151 source_file: str, 

152 name: str, 

153 annotations: dict[str, str], 

154 labels: dict[str, str], 

155 image: Optional[str] = None, 

156 ports: Optional[list[Port]] = None, 

157 networks: Optional[list[str]] = None, 

158 volumes: Optional[list[Volume]] = None, 

159 depends_on: Optional[dict[str, spec.DependsOn]] = None, 

160 links: Optional[list[str]] = None, 

161 extends: Optional[Extends] = None, 

162 cgroup_parent: Optional[str] = None, 

163 container_name: Optional[str] = None, 

164 devices: Optional[list[Device]] = None, 

165 env_file: Optional[dict[str, EnvFileInfo]] = None, 

166 expose: Optional[list[str]] = None, 

167 profiles: Optional[list[str]] = None, 

168 ) -> None: 

169 self.name = name 

170 self.image = image 

171 self.ports = opt_to_arr(ports) 

172 self.networks = opt_to_arr(networks) 

173 self.volumes = opt_to_arr(volumes) 

174 self.depends_on = opt_to_dict(depends_on) 

175 self.links = opt_to_arr(links) 

176 self.extends = extends 

177 self.cgroup_parent = cgroup_parent 

178 self.container_name = container_name 

179 self.devices = opt_to_arr(devices) 

180 self.env_file = opt_to_dict(env_file) 

181 self.expose = opt_to_arr(expose) 

182 self.profiles = opt_to_arr(profiles) 

183 self.labels = labels 

184 self.annotations = annotations 

185 self.source_files: list[str] = [source_file] 

186 

187 def merge(self, other: "Service") -> None: 

188 """ 

189 Merge a service with a pre-existing definition. 

190 

191 All attributes of other parameter will replace or merge existing ones when set. 

192 """ 

193 if other.image: 

194 self.image = other.image 

195 if other.name: 

196 self.name = other.name 

197 

198 

199class Network: 

200 # pylint: disable=too-many-instance-attributes 

201 def __init__(self, source_file: str, network: spec.Network) -> None: 

202 self.name: Optional[str] = None 

203 self.driver: Optional[str] = None 

204 self.driver_opts: Optional[dict[str, Union[str, float]]] = None 

205 self.ipam: Optional[spec.Ipam] = None 

206 self.external = _parse_external(network.external) 

207 self.internal: bool = False if network.internal is None else network.internal 

208 self.enable_ipv6: Optional[bool] = network.enable_ipv6 

209 self.attachable: Optional[bool] = network.attachable 

210 self.labels: dict[str, str] = parse_list_of_dict_of_tuples(network.labels) 

211 self.source_file = [source_file] 

212 

213 

214class Secret: 

215 # pylint: disable=too-many-instance-attributes 

216 def __init__(self, source_file: str, secret: spec.Secret) -> None: 

217 self.name: Optional[str] = secret.name 

218 self.environment: Optional[str] = secret.environment 

219 self.file: Optional[str] = secret.file 

220 self.external = _parse_external(secret.external) 

221 self.labels = parse_list_of_dict_of_tuples(secret.labels) 

222 self.driver: Optional[str] = secret.driver 

223 self.driver_opts: dict[str, Union[str, float]] = ( 

224 secret.driver_opts if secret.driver_opts else {} 

225 ) 

226 self.template_driver: Optional[str] = secret.template_driver 

227 self.source_file = [source_file] 

228 

229 

230class Config: 

231 # pylint: disable=too-many-instance-attributes 

232 def __init__(self, source_file: str, config: spec.Config) -> None: 

233 self.name: Optional[str] = config.name 

234 self.content: Optional[str] = config.content 

235 self.environment: Optional[str] = config.environment 

236 self.file: Optional[str] = config.file 

237 self.external = _parse_external(config.external) 

238 self.labels = parse_list_of_dict_of_tuples(config.labels) 

239 self.template_driver: Optional[str] = config.template_driver 

240 self.source_file = [source_file] 

241 

242 

243class RootVolume: 

244 def __init__(self, source_file, volume=spec.Volume) -> None: 

245 if volume is None: 

246 volume = spec.Volume() 

247 self.name: Optional[str] = volume.name 

248 self.driver = volume.driver 

249 self.driver_opts = parse_list_of_dict_of_tuples(volume.driver_opts) 

250 self.external = _parse_external(volume.external) 

251 self.labels = parse_list_of_dict_of_tuples(volume.labels) 

252 self.source_file = [source_file]