Skip to content

Reference

odooghost.config

ContextConfig

Bases: BaseModel

Context config holds configuration file

Source code in odooghost/config/app.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
class ContextConfig(BaseModel):
    """
    Context config holds configuration file
    """

    version: str
    """
    OdooGhost version
    """
    working_dir: Path
    """
    Working directory
    """

version: str instance-attribute

OdooGhost version

working_dir: Path instance-attribute

Working directory

OdooStackConfig

Bases: StackServiceConfig

Odoo stack configuration

Source code in odooghost/config/service.py 
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
80
81
82
83
84
85
86
87
88
89
class OdooStackConfig(StackServiceConfig):
    """
    Odoo stack configuration
    """

    version: float
    """
    Odoo version
    """
    cmdline: t.Optional[str] = None
    """
    Odoo-bin cmdline
    """
    addons: t.List[_addons.AddonsConfig] = []
    """
    Odoo addons configurations
    """
    dependencies: dependency.DependenciesConfig = dependency.DependenciesConfig()
    """
    Odoo dependencies configurations
    """

    @field_validator("version")
    @classmethod
    def validate_versîon(cls, v: float) -> float:
        """
        Validate supported Odoo version

        Raises:
            ValueError: When provided version is not supported

        Returns:
            float: Odoo version
        """
        if v not in (11.0, 12.0, 13.0, 14.0, 15.0, 16.0, 17.0, 18.0):
            raise ValueError(f"Unsuported Odoo version {v}")
        return v

addons: t.List[_addons.AddonsConfig] = [] class-attribute instance-attribute

Odoo addons configurations

cmdline: t.Optional[str] = None class-attribute instance-attribute

Odoo-bin cmdline

dependencies: dependency.DependenciesConfig = dependency.DependenciesConfig() class-attribute instance-attribute

Odoo dependencies configurations

version: float instance-attribute

Odoo version

validate_versîon(v) classmethod

Validate supported Odoo version

Raises:

Type Description
ValueError

When provided version is not supported

Returns:

Name Type Description
float float

Odoo version
Source code in odooghost/config/service.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
@field_validator("version")
@classmethod
def validate_versîon(cls, v: float) -> float:
    """
    Validate supported Odoo version

    Raises:
        ValueError: When provided version is not supported

    Returns:
        float: Odoo version
    """
    if v not in (11.0, 12.0, 13.0, 14.0, 15.0, 16.0, 17.0, 18.0):
        raise ValueError(f"Unsuported Odoo version {v}")
    return v

PostgresStackConfig

Bases: StackServiceConfig

Postgres stack configuration holds database configuration It support both remote and local databse

Source code in odooghost/config/service.py 
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
class PostgresStackConfig(StackServiceConfig):
    """
    Postgres stack configuration holds database configuration
    It support both remote and local databse
    """

    type: t.Literal["local", "remote"]
    """
    Type of database config
    """
    version: int
    """
    Database version
    """
    host: t.Optional[str] = None
    """
    Database hostname
    """
    user: t.Optional[str] = None
    """
    Database user
    """
    db: t.Optional[str] = "postgres"
    """
    Database template (only availible in local type)
    """
    password: t.Optional[str] = None
    """
    Database user password
    """

db: t.Optional[str] = 'postgres' class-attribute instance-attribute

Database template (only availible in local type)

host: t.Optional[str] = None class-attribute instance-attribute

Database hostname

password: t.Optional[str] = None class-attribute instance-attribute

Database user password

type: t.Literal['local', 'remote'] instance-attribute

Type of database config

user: t.Optional[str] = None class-attribute instance-attribute

Database user

version: int instance-attribute

Database version

StackConfig

Bases: BaseModel

Stack configuration

Source code in odooghost/config/stack.py 
 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
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
class StackConfig(BaseModel):
    """
    Stack configuration
    """

    name: str
    """
    Name of stack
    """
    services: service.StackServicesConfig
    """
    Services of stack
    """
    network: StackNetworkConfig = StackNetworkConfig()
    """
    Network config
    """

    @field_validator("name")
    @classmethod
    def validate_name(cls, v: str) -> str:
        """
        Validate stack name

        Args:
            v (str): Stack name

        Raises:
            ValueError: When stack name is not valid

        Returns:
            str: Stack name
        """
        if " " in v or not re.match(r"^[\w-]+$", v):
            raise ValueError("Stack name must not contain spaces or special characters")
        return v

    @classmethod
    def from_file(cls, file_path: Path) -> "StackConfig":
        """
        Return a StackConfig instance from JSON/YAML file config

        Args:
            file_path (Path): file path

        Raises:
            RuntimeError: when the file does not exists

        Returns:
            StackConfig: StackConfig instance
        """
        if not file_path.exists():
            # TODO replace this error
            raise RuntimeError("File does not exist")
        data = {}
        with open(file_path.as_posix(), "r") as stream:
            if file_path.name.endswith(".json"):
                data = json.load(fp=stream)
            elif file_path.name.endswith(".yml") or file_path.name.endswith(".yaml"):
                data = yaml.safe_load(stream=stream)
            else:
                raise exceptions.StackConfigError("Unsupported file format")
        return cls(**data)

    def get_service_hostname(self, service: str) -> str:
        """
        Get given service name regatding netowrk.
        We do prefix the service name with the stack name
        if the stack network is shared with other.
        This is done to allow running multiple stack's at
        the same time with the same network

        Args:
            service (str): service name

        Returns:
            str: name of the given service
        """
        return (
            f"{self.name.lower()}-{service}"
            if self.network.mode == "shared"
            else service
        )

    def get_network_name(self) -> str:
        """
        Get netowkr name regarding network mode
        T

        Returns:
            str: Stack netowrk name
        """
        return constant.COMMON_NETWORK_NAME or f"{constant.LABEL_NAME}_{self.name}"

name: str instance-attribute

Name of stack

network: StackNetworkConfig = StackNetworkConfig() class-attribute instance-attribute

Network config

services: service.StackServicesConfig instance-attribute

Services of stack

from_file(file_path) classmethod

Return a StackConfig instance from JSON/YAML file config

Parameters:

Name Type Description Default
file_path Path

file path

 required

Raises:

Type Description
RuntimeError

when the file does not exists

Returns:

Name Type Description
StackConfig StackConfig

StackConfig instance
Source code in odooghost/config/stack.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
@classmethod
def from_file(cls, file_path: Path) -> "StackConfig":
    """
    Return a StackConfig instance from JSON/YAML file config

    Args:
        file_path (Path): file path

    Raises:
        RuntimeError: when the file does not exists

    Returns:
        StackConfig: StackConfig instance
    """
    if not file_path.exists():
        # TODO replace this error
        raise RuntimeError("File does not exist")
    data = {}
    with open(file_path.as_posix(), "r") as stream:
        if file_path.name.endswith(".json"):
            data = json.load(fp=stream)
        elif file_path.name.endswith(".yml") or file_path.name.endswith(".yaml"):
            data = yaml.safe_load(stream=stream)
        else:
            raise exceptions.StackConfigError("Unsupported file format")
    return cls(**data)

get_network_name()

Get netowkr name regarding network mode T

Returns:

Name Type Description
str str

Stack netowrk name
Source code in odooghost/config/stack.py 
106
107
108
109
110
111
112
113
114
def get_network_name(self) -> str:
    """
    Get netowkr name regarding network mode
    T

    Returns:
        str: Stack netowrk name
    """
    return constant.COMMON_NETWORK_NAME or f"{constant.LABEL_NAME}_{self.name}"

get_service_hostname(service)

Get given service name regatding netowrk. We do prefix the service name with the stack name if the stack network is shared with other. This is done to allow running multiple stack's at the same time with the same network

Parameters:

Name Type Description Default
service str

service name

 required

Returns:

Name Type Description
str str

name of the given service
Source code in odooghost/config/stack.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def get_service_hostname(self, service: str) -> str:
    """
    Get given service name regatding netowrk.
    We do prefix the service name with the stack name
    if the stack network is shared with other.
    This is done to allow running multiple stack's at
    the same time with the same network

    Args:
        service (str): service name

    Returns:
        str: name of the given service
    """
    return (
        f"{self.name.lower()}-{service}"
        if self.network.mode == "shared"
        else service
    )

validate_name(v) classmethod

Validate stack name

Parameters:

Name Type Description Default
v str

Stack name

 required

Raises:

Type Description
ValueError

When stack name is not valid

Returns:

Name Type Description
str str

Stack name
Source code in odooghost/config/stack.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
@field_validator("name")
@classmethod
def validate_name(cls, v: str) -> str:
    """
    Validate stack name

    Args:
        v (str): Stack name

    Raises:
        ValueError: When stack name is not valid

    Returns:
        str: Stack name
    """
    if " " in v or not re.match(r"^[\w-]+$", v):
        raise ValueError("Stack name must not contain spaces or special characters")
    return v

StackServiceConfig

Bases: BaseModel, ABC

Abstract config for stack services

Source code in odooghost/config/service.py 
10
11
12
13
14
15
16
17
18
class StackServiceConfig(BaseModel, abc.ABC):
    """
    Abstract config for stack services
    """

    service_port: t.Optional[int] = None
    """
    Map local port to container sercice port
    """

service_port: t.Optional[int] = None class-attribute instance-attribute

Map local port to container sercice port

odooghost.stack

Stack

Stack manage differents Odoo stacks regarding it's config

Source code in odooghost/stack.py 
 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
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
class Stack:
    """
    Stack manage differents Odoo stacks regarding it's config
    """

    def __init__(self, config: "config.StackConfig") -> None:
        self._config = config
        self._services: t.Dict[str, t.Type["BaseService"]] = dict(
            db=db.DbService(stack_config=config),
            odoo=odoo.OdooService(stack_config=config),
        )
        if config.services.mail:
            self._services.update(dict(mail=mail.MailService(stack_config=config)))

    def _check_state(self) -> StackState:
        """
        Check Stack state

        Returns:
            StackState: return current state
        """
        if self.name not in ctx.stacks:
            return StackState.NONE
        # TODO implement partial state
        return StackState.READY

    def _ensure_exists(func: t.Callable) -> t.Callable:
        """
        Ensure Stack exists

        Args:
            func (t.Callable): function to call

        Raises:
            StackNotFoundError: When Stack does not exists

        Returns:
            t.Callable: wrapped function
        """

        @wraps(func)
        def inner(self: "Stack", *args, **kwargs) -> t.Any:
            if not self.exists:
                raise StackNotFoundError(f"Stack {self.name} does not exists !")
            return func(self, *args, **kwargs)

        return inner

    @classmethod
    def from_file(cls, file_path: Path) -> "Stack":
        """
        Instanciate Stack from file

        File could be both YAML and JSON

        Args:
            file_path (Path): stack config file path

        Returns:
            Stack: Stack instance
        """
        return cls(config=config.StackConfig.from_file(file_path=file_path))

    @classmethod
    def from_name(cls, name: str) -> "Stack":
        """
        Instanciate Stack from name
        Stack config will be searched from Context

        Args:
            name (str): Stack name

        Returns:
            Stack: Stack instance
        """
        return cls(config=ctx.stacks.get(stack_name=name))

    @classmethod
    def count(cls) -> int:
        """
        Count all stacks in context

        Returns:
            int: stack count
        """
        return len(ctx.stacks)

    @classmethod
    def list(cls, running: bool = False) -> t.Generator["Stack", None, None]:
        """
        List all stacks

        Yields:
            Srack: Stack instance
        """
        # TODO implment running stack only
        if running:
            for stack_name in set(
                map(
                    lambda container: container.stack,
                    Container.search(
                        filters={
                            "label": labels_as_list(
                                {
                                    constant.LABEL_NAME: "true",
                                }
                            )
                        }
                    ),
                )
            ):
                yield cls.from_name(name=stack_name)

        else:
            for stack_config in ctx.stacks:
                yield cls(config=stack_config)

    def labels(self, one_off: OneOffFilter = OneOffFilter.exclude) -> Labels:
        """
        Get Stack labels

        Returns:
            Labels: Labels as dict
        """
        labels = {
            constant.LABEL_NAME: "true",
            constant.LABEL_STACKNAME: self.name,
        }
        OneOffFilter.update_labels(value=one_off, labels=labels)
        return labels

    def services(self) -> t.List[t.Type["BaseService"]]:
        return list(self._services.values())

    def get_service(self, name: str) -> t.Type["BaseService"]:
        try:
            service = self._services[name]
        except KeyError:
            # TODO make exception
            raise Exception
        return service

    def containers(
        self,
        filters: t.Optional[Filters] = None,
        labels: t.Optional[Labels] = None,
        stopped: bool = False,
        one_off: OneOffFilter = OneOffFilter.exclude,
    ) -> t.List[Container]:
        """
        Get Stack containers

        Args:
            filters (t.Optional[Filters], optional): Search filters. Defaults to None.
            labels (t.Optional[Labels], optional): Additionnal search labels. Defaults to None.
            stopped (bool, optional): Get stopped containers. Defaults to False.

        Returns:
            t.List[Container]: Container list
        """
        if filters is None:
            filters = {}
        filters.update(
            {
                "label": labels_as_list(self.labels(one_off=one_off))
                + (labels_as_list(labels) if labels else [])
            }
        )
        return Container.search(filters=filters, stopped=stopped)

    def create(
        self, force: bool = False, do_pull: bool = True, ensure_addons: bool = True
    ) -> None:
        """
        Create Stack

        Args:
            force (bool, optional): Force recreate of dangling containers. Defaults to False.
            do_pull (bool, optional): Pull base images. Defaults to True.
            ensure_addons (bool, optional): Ensure Odoo addons. Defaults to True.

        Raises:
            StackAlreadyExistsError: When Stack alreary exists
        """
        if self.exists:
            raise StackAlreadyExistsError(f"Stack {self.name} already exists !")
        logger.info(f"Creating Stack {self.name} ...")
        # TODO allow custom network
        ctx.ensure_common_network()
        for service in self.services():
            service.create(force=force, do_pull=do_pull, ensure_addons=ensure_addons)

        ctx.stacks.create(config=self._config)
        logger.info(f"Created Stack {self.name} !")

    @_ensure_exists
    def drop(self, volumes: bool = False) -> None:
        """
        Drop Stack

        Args:
            volumes (bool, optional): Drop volumes. Defaults to False.

        Raises:
            StackNotFoundError: When Stack does not exists
        """
        logger.info(f"Dropping Stack {self.name} ...")
        for service in self.services():
            service.drop(volumes=volumes)
        ctx.stacks.drop(stack_name=self.name)
        logger.info(f"Dropped Stack {self.name} !")

    @_ensure_exists
    def pull(self) -> None:
        """
        Pull Stack
        """
        logger.info(f"Pulling Stack {self.name} ...")
        for service in self.services():
            service.pull()
        logger.info(f"Pulled Stack {self.name} !")

    @_ensure_exists
    def update(self, do_pull: bool = False) -> None:
        """
        Update Stack
        """
        logger.info(f"Updating Stack {self.name} ...")
        for service in self.services():
            if do_pull:
                service.pull()
            service.update()
        ctx.stacks.update(config=self._config)
        logger.info(f"Updated Stack {self.name} !")

    @_ensure_exists
    def start(self) -> None:
        """
        Start Stack

        Raises:
            StackNotFoundError: When Stack does not exists
        """
        containers = self.containers(stopped=True)
        if not len(containers):
            logger.warning("No container to start !")
            return
        for container in containers:
            logger.info(f"Starting container {container.name}")
            container.start()

    @_ensure_exists
    def stop(self, timeout: int = 10, wait: bool = False) -> None:
        """
        Stop Stack

        Args:
            timeout (int, optional): timeout before sending SIGKILL. Defaults to 10.

        Raises:
            StackNotFoundError: When stack does not exists
        """
        containers = self.containers()
        if not len(containers):
            logger.warning("No container to stop !")
            return
        for container in containers:
            logger.info(f"Stopping container {container.name}")
            container.stop(timeout=timeout)
        if wait:
            logger.info("Waiting for containers to stop")
            for container in containers:
                container.wait()

    @_ensure_exists
    def restart(self, timeout: int = 10) -> None:
        """
        Restart Stack

        Args:
            timeout (int, optional): timeout before sending SIGKILL. Defaults to 10.

        Raises:
            StackNotFoundError: When stack does not exists
        """
        containers = self.containers()
        if not len(containers):
            logger.warning("No container to restart !")
            return
        for container in containers:
            logger.info(f"Restarting container {container.name}")
            container.restart(timeout=timeout)

    @property
    def name(self) -> str:
        """
        Return name of stack

        Returns:
            str: Stack name
        """
        return self._config.name

    @property
    def state(self) -> StackState:
        """
        Check state and return it

        Returns:
            StackState: Current Stack state
        """
        return self._check_state()

    @property
    def exists(self) -> bool:
        """
        Check if stack already exists

        Returns:
            bool
        """
        return self.state != StackState.NONE

    @property
    def id(self) -> str:
        return get_hash(self.name)

    def __repr__(self):
        """
        Stack repr
        """
        return f"<Stack: {self.name}>"

    def __eq__(self, other: "Stack") -> bool:
        """
        Check if Stack equal other Stack

        Args:
            other (Stack): Other Stack instance

        Returns:
            bool:
        """
        if not isinstance(other, self.__class__):
            return False
        return self.name == other.name

exists: bool property

Check if stack already exists

Returns:

Type Description
bool

bool

name: str property

Return name of stack

Returns:

Name Type Description
str str

Stack name

state: StackState property

Check state and return it

Returns:

Name Type Description
StackState StackState

Current Stack state

__eq__(other)

Check if Stack equal other Stack

Parameters:

Name Type Description Default
other Stack

Other Stack instance

 required

Returns:

Name Type Description
bool bool
Source code in odooghost/stack.py 
365
366
367
368
369
370
371
372
373
374
375
376
377
def __eq__(self, other: "Stack") -> bool:
    """
    Check if Stack equal other Stack

    Args:
        other (Stack): Other Stack instance

    Returns:
        bool:
    """
    if not isinstance(other, self.__class__):
        return False
    return self.name == other.name

__repr__()

Stack repr

Source code in odooghost/stack.py 
359
360
361
362
363
def __repr__(self):
    """
    Stack repr
    """
    return f"<Stack: {self.name}>"

containers(filters=None, labels=None, stopped=False, one_off=OneOffFilter.exclude)

Get Stack containers

Parameters:

Name Type Description Default
filters Optional[Filters]

Search filters. Defaults to None.

 None
labels Optional[Labels]

Additionnal search labels. Defaults to None.

 None
stopped bool

Get stopped containers. Defaults to False.

 False

Returns:

Type Description
List[Container]

t.List[Container]: Container list
Source code in odooghost/stack.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
def containers(
    self,
    filters: t.Optional[Filters] = None,
    labels: t.Optional[Labels] = None,
    stopped: bool = False,
    one_off: OneOffFilter = OneOffFilter.exclude,
) -> t.List[Container]:
    """
    Get Stack containers

    Args:
        filters (t.Optional[Filters], optional): Search filters. Defaults to None.
        labels (t.Optional[Labels], optional): Additionnal search labels. Defaults to None.
        stopped (bool, optional): Get stopped containers. Defaults to False.

    Returns:
        t.List[Container]: Container list
    """
    if filters is None:
        filters = {}
    filters.update(
        {
            "label": labels_as_list(self.labels(one_off=one_off))
            + (labels_as_list(labels) if labels else [])
        }
    )
    return Container.search(filters=filters, stopped=stopped)

count() classmethod

Count all stacks in context

Returns:

Name Type Description
int int

stack count
Source code in odooghost/stack.py 
109
110
111
112
113
114
115
116
117
@classmethod
def count(cls) -> int:
    """
    Count all stacks in context

    Returns:
        int: stack count
    """
    return len(ctx.stacks)

create(force=False, do_pull=True, ensure_addons=True)

Create Stack

Parameters:

Name Type Description Default
force bool

Force recreate of dangling containers. Defaults to False.

 False
do_pull bool

Pull base images. Defaults to True.

 True
ensure_addons bool

Ensure Odoo addons. Defaults to True.

 True

Raises:

Type Description
StackAlreadyExistsError

When Stack alreary exists
Source code in odooghost/stack.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
def create(
    self, force: bool = False, do_pull: bool = True, ensure_addons: bool = True
) -> None:
    """
    Create Stack

    Args:
        force (bool, optional): Force recreate of dangling containers. Defaults to False.
        do_pull (bool, optional): Pull base images. Defaults to True.
        ensure_addons (bool, optional): Ensure Odoo addons. Defaults to True.

    Raises:
        StackAlreadyExistsError: When Stack alreary exists
    """
    if self.exists:
        raise StackAlreadyExistsError(f"Stack {self.name} already exists !")
    logger.info(f"Creating Stack {self.name} ...")
    # TODO allow custom network
    ctx.ensure_common_network()
    for service in self.services():
        service.create(force=force, do_pull=do_pull, ensure_addons=ensure_addons)

    ctx.stacks.create(config=self._config)
    logger.info(f"Created Stack {self.name} !")

drop(volumes=False)

Drop Stack

Parameters:

Name Type Description Default
volumes bool

Drop volumes. Defaults to False.

 False

Raises:

Type Description
StackNotFoundError

When Stack does not exists
Source code in odooghost/stack.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
@_ensure_exists
def drop(self, volumes: bool = False) -> None:
    """
    Drop Stack

    Args:
        volumes (bool, optional): Drop volumes. Defaults to False.

    Raises:
        StackNotFoundError: When Stack does not exists
    """
    logger.info(f"Dropping Stack {self.name} ...")
    for service in self.services():
        service.drop(volumes=volumes)
    ctx.stacks.drop(stack_name=self.name)
    logger.info(f"Dropped Stack {self.name} !")

from_file(file_path) classmethod

Instanciate Stack from file

File could be both YAML and JSON

Parameters:

Name Type Description Default
file_path Path

stack config file path

 required

Returns:

Name Type Description
Stack Stack

Stack instance
Source code in odooghost/stack.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
@classmethod
def from_file(cls, file_path: Path) -> "Stack":
    """
    Instanciate Stack from file

    File could be both YAML and JSON

    Args:
        file_path (Path): stack config file path

    Returns:
        Stack: Stack instance
    """
    return cls(config=config.StackConfig.from_file(file_path=file_path))

from_name(name) classmethod

Instanciate Stack from name Stack config will be searched from Context

Parameters:

Name Type Description Default
name str

Stack name

 required

Returns:

Name Type Description
Stack Stack

Stack instance
Source code in odooghost/stack.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@classmethod
def from_name(cls, name: str) -> "Stack":
    """
    Instanciate Stack from name
    Stack config will be searched from Context

    Args:
        name (str): Stack name

    Returns:
        Stack: Stack instance
    """
    return cls(config=ctx.stacks.get(stack_name=name))

labels(one_off=OneOffFilter.exclude)

Get Stack labels

Returns:

Name Type Description
Labels Labels

Labels as dict
Source code in odooghost/stack.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
def labels(self, one_off: OneOffFilter = OneOffFilter.exclude) -> Labels:
    """
    Get Stack labels

    Returns:
        Labels: Labels as dict
    """
    labels = {
        constant.LABEL_NAME: "true",
        constant.LABEL_STACKNAME: self.name,
    }
    OneOffFilter.update_labels(value=one_off, labels=labels)
    return labels

list(running=False) classmethod

List all stacks

Yields:

Name Type Description
Srack Stack

Stack instance
Source code in odooghost/stack.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
@classmethod
def list(cls, running: bool = False) -> t.Generator["Stack", None, None]:
    """
    List all stacks

    Yields:
        Srack: Stack instance
    """
    # TODO implment running stack only
    if running:
        for stack_name in set(
            map(
                lambda container: container.stack,
                Container.search(
                    filters={
                        "label": labels_as_list(
                            {
                                constant.LABEL_NAME: "true",
                            }
                        )
                    }
                ),
            )
        ):
            yield cls.from_name(name=stack_name)

    else:
        for stack_config in ctx.stacks:
            yield cls(config=stack_config)

pull()

Pull Stack

Source code in odooghost/stack.py 
244
245
246
247
248
249
250
251
252
@_ensure_exists
def pull(self) -> None:
    """
    Pull Stack
    """
    logger.info(f"Pulling Stack {self.name} ...")
    for service in self.services():
        service.pull()
    logger.info(f"Pulled Stack {self.name} !")

restart(timeout=10)

Restart Stack

Parameters:

Name Type Description Default
timeout int

timeout before sending SIGKILL. Defaults to 10.

 10

Raises:

Type Description
StackNotFoundError

When stack does not exists
Source code in odooghost/stack.py 
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
@_ensure_exists
def restart(self, timeout: int = 10) -> None:
    """
    Restart Stack

    Args:
        timeout (int, optional): timeout before sending SIGKILL. Defaults to 10.

    Raises:
        StackNotFoundError: When stack does not exists
    """
    containers = self.containers()
    if not len(containers):
        logger.warning("No container to restart !")
        return
    for container in containers:
        logger.info(f"Restarting container {container.name}")
        container.restart(timeout=timeout)

start()

Start Stack

Raises:

Type Description
StackNotFoundError

When Stack does not exists
Source code in odooghost/stack.py 
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
@_ensure_exists
def start(self) -> None:
    """
    Start Stack

    Raises:
        StackNotFoundError: When Stack does not exists
    """
    containers = self.containers(stopped=True)
    if not len(containers):
        logger.warning("No container to start !")
        return
    for container in containers:
        logger.info(f"Starting container {container.name}")
        container.start()

stop(timeout=10, wait=False)

Stop Stack

Parameters:

Name Type Description Default
timeout int

timeout before sending SIGKILL. Defaults to 10.

 10

Raises:

Type Description
StackNotFoundError

When stack does not exists
Source code in odooghost/stack.py 
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
@_ensure_exists
def stop(self, timeout: int = 10, wait: bool = False) -> None:
    """
    Stop Stack

    Args:
        timeout (int, optional): timeout before sending SIGKILL. Defaults to 10.

    Raises:
        StackNotFoundError: When stack does not exists
    """
    containers = self.containers()
    if not len(containers):
        logger.warning("No container to stop !")
        return
    for container in containers:
        logger.info(f"Stopping container {container.name}")
        container.stop(timeout=timeout)
    if wait:
        logger.info("Waiting for containers to stop")
        for container in containers:
            container.wait()

update(do_pull=False)

Update Stack

Source code in odooghost/stack.py 
254
255
256
257
258
259
260
261
262
263
264
265
@_ensure_exists
def update(self, do_pull: bool = False) -> None:
    """
    Update Stack
    """
    logger.info(f"Updating Stack {self.name} ...")
    for service in self.services():
        if do_pull:
            service.pull()
        service.update()
    ctx.stacks.update(config=self._config)
    logger.info(f"Updated Stack {self.name} !")

StackState

Bases: Enum

StackState obviously holds StackState

Source code in odooghost/stack.py 
21
22
23
24
25
26
27
28
class StackState(enum.Enum):
    """
    StackState obviously holds StackState
    """

    NONE: int = 0
    PARTIAL: int = 1
    READY: int = 2

odooghost.services

odooghost.context

Context

Context holds contextual data for OdooGhost

Source code in odooghost/context.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
class Context:
    """
    Context holds contextual data for OdooGhost
    """

    def __init__(self) -> None:
        self._app_dir = constant.APP_DIR
        self._config_path = self._app_dir / "config.yml"
        self._stack_ctx = StackContext(self._app_dir / "stacks")
        self._data_dir = self._app_dir / "data"
        self._plugins_dir = self._app_dir / "plugins"
        self._config: t.Optional[ContextConfig] = None
        self._docker_client: t.Optional[docker.DockerClient] = None
        self._init = False
        self.initialize()

    def check_setup_state(self) -> bool:
        """
        Check setup status

        Returns:
            bool
        """
        return self._app_dir.exists()

    def initialize(self) -> None:
        """
        Initialize context
        """
        if self.check_setup_state():
            with open(self._config_path.as_posix(), "r") as stream:
                self._config = ContextConfig(**yaml.safe_load(stream=stream))
            self._init = True

    def setup(self, version: str, working_dir: Path) -> None:
        """
        Setup OdooGhost

        Args:
            version (str): OdooGhost version
            working_dir (Path): working directory

        Raises:
            exceptions.ContextAlreadySetupError: Already setup
        """
        if self.check_setup_state():
            raise exceptions.ContextAlreadySetupError("App already setup !")

        # TODO handle OSError
        for _dir in (
            self._app_dir,
            self._stack_ctx._working_dir,
            self._data_dir,
            self._plugins_dir,
        ):
            _dir.mkdir()
        config_data = dict(
            version=version, working_dir=working_dir.resolve().as_posix()
        )
        with open(self._config_path.as_posix(), "w") as stream:
            yaml.safe_dump(config_data, stream=stream)

        self.initialize()

    def create_common_network(self) -> None:
        """
        Create common Docker network for stacks

        Raises:
            exceptions.CommonNetworkEnsureError: When create fail
        """
        try:
            self.docker.networks.create(
                name=constant.COMMON_NETWORK_NAME,
                driver="bridge",
                check_duplicate=True,
                attachable=True,
                scope="local",
            )
        except APIError:
            raise exceptions.CommonNetworkEnsureError("Failed to create common network")

    def ensure_common_network(self) -> None:
        """
        Ensure common Docker network

        Raises:
            exceptions.CommonNetworkEnsureError: When ensure fail
        """
        try:
            self.docker.networks.get(constant.COMMON_NETWORK_NAME)
        except NotFound:
            self.create_common_network()
        except APIError:
            raise exceptions.CommonNetworkEnsureError("Failed to ensure common network")

    def get_build_context_path(self) -> Path:
        """
        Get build context path

        Returns:
            Path: Path to build context
        """
        return Path("/tmp/odooghost")  # nosec B108

    @property
    def docker(self) -> "docker.DockerClient":
        """
        Lazyily return Docker client

        Returns:
            docker.DockerClient: Docker client instance
        """
        if not self._docker_client:
            self._docker_client = docker.from_env()
        return self._docker_client

    @property
    def config(self) -> ContextConfig:
        """
        Get context config

        Raises:
            RuntimeError: when context was not initialized

        Returns:
            ContextConfig: context config
        """
        if not self._init:
            raise RuntimeError("Can not get config before initialize has been done")
        return self._config

    @property
    def stacks(self) -> StackContext:
        if not self._init:
            raise RuntimeError(
                "Can not get stack config manager before initialize has been done"
            )
        return self._stack_ctx

config: ContextConfig property

Get context config

Raises:

Type Description
RuntimeError

when context was not initialized

Returns:

Name Type Description
ContextConfig ContextConfig

context config

docker: docker.DockerClient property

Lazyily return Docker client

Returns:

Type Description
DockerClient

docker.DockerClient: Docker client instance

check_setup_state()

Check setup status

Returns:

Type Description
bool

bool
Source code in odooghost/context.py 
137
138
139
140
141
142
143
144
def check_setup_state(self) -> bool:
    """
    Check setup status

    Returns:
        bool
    """
    return self._app_dir.exists()

create_common_network()

Create common Docker network for stacks

Raises:

Type Description
CommonNetworkEnsureError

When create fail
Source code in odooghost/context.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
def create_common_network(self) -> None:
    """
    Create common Docker network for stacks

    Raises:
        exceptions.CommonNetworkEnsureError: When create fail
    """
    try:
        self.docker.networks.create(
            name=constant.COMMON_NETWORK_NAME,
            driver="bridge",
            check_duplicate=True,
            attachable=True,
            scope="local",
        )
    except APIError:
        raise exceptions.CommonNetworkEnsureError("Failed to create common network")

ensure_common_network()

Ensure common Docker network

Raises:

Type Description
CommonNetworkEnsureError

When ensure fail
Source code in odooghost/context.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
def ensure_common_network(self) -> None:
    """
    Ensure common Docker network

    Raises:
        exceptions.CommonNetworkEnsureError: When ensure fail
    """
    try:
        self.docker.networks.get(constant.COMMON_NETWORK_NAME)
    except NotFound:
        self.create_common_network()
    except APIError:
        raise exceptions.CommonNetworkEnsureError("Failed to ensure common network")

get_build_context_path()

Get build context path

Returns:

Name Type Description
Path Path

Path to build context
Source code in odooghost/context.py 
217
218
219
220
221
222
223
224
def get_build_context_path(self) -> Path:
    """
    Get build context path

    Returns:
        Path: Path to build context
    """
    return Path("/tmp/odooghost")  # nosec B108

initialize()

Initialize context

Source code in odooghost/context.py 
146
147
148
149
150
151
152
153
def initialize(self) -> None:
    """
    Initialize context
    """
    if self.check_setup_state():
        with open(self._config_path.as_posix(), "r") as stream:
            self._config = ContextConfig(**yaml.safe_load(stream=stream))
        self._init = True

setup(version, working_dir)

Setup OdooGhost

Parameters:

Name Type Description Default
version str

OdooGhost version

 required
working_dir Path

working directory

 required

Raises:

Type Description
ContextAlreadySetupError

Already setup
Source code in odooghost/context.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def setup(self, version: str, working_dir: Path) -> None:
    """
    Setup OdooGhost

    Args:
        version (str): OdooGhost version
        working_dir (Path): working directory

    Raises:
        exceptions.ContextAlreadySetupError: Already setup
    """
    if self.check_setup_state():
        raise exceptions.ContextAlreadySetupError("App already setup !")

    # TODO handle OSError
    for _dir in (
        self._app_dir,
        self._stack_ctx._working_dir,
        self._data_dir,
        self._plugins_dir,
    ):
        _dir.mkdir()
    config_data = dict(
        version=version, working_dir=working_dir.resolve().as_posix()
    )
    with open(self._config_path.as_posix(), "w") as stream:
        yaml.safe_dump(config_data, stream=stream)

    self.initialize()

StackContext

Source code in odooghost/context.py 
 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
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
class StackContext:
    def __init__(self, working_dir: Path) -> None:
        self._working_dir = working_dir

    def _write(self, config: StackConfig) -> None:
        with open(self.get_path(config.name), "w") as stream:
            json.dump(config.model_dump(), stream)

    def get_path(self, stack_name: str) -> Path:
        """
        Get Stack config path

        Args:
            stack_name (str): name of stack

        Returns:
            Path: Stack config path
        """
        return self._working_dir / f"{stack_name}.json"

    def get(self, stack_name: str) -> StackConfig:
        """
        Get StackConfig

        Args:
            stack_name (str): name of stack

        Returns:
            StackConfig: Stack config instance
        """
        if stack_name not in self:
            raise exceptions.StackNotFoundError(
                f"Stack {stack_name} config file doest not exists"
            )
        return StackConfig.from_file(file_path=self.get_path(stack_name=stack_name))

    def create(self, config: StackConfig) -> None:
        """
        Create StackConfig file in context

        Args:
            config (StackConfig): Stack config

        Raises:
            exceptions.StackAlreadyExistsError: When stack config file exists
        """
        if config in self:
            raise exceptions.StackAlreadyExistsError(
                f"Stack {config.name} already exists"
            )
        self._write(config=config)

    def update(self, config: StackConfig) -> None:
        """
        Update StackConfig file in context

        Args:
            config (StackConfig): Stack config

        Raises:
            exceptions.StackNotFoundError: When stack config file does not exists
        """
        if config not in self:
            raise exceptions.StackNotFoundError(f"Stack {config.name} not found")
        self._write(config=config)

    def drop(self, stack_name: str) -> None:
        """
        Drop Stack from context

        Args:
            stack_name (str): name of stack
        """
        if stack_name not in self:
            raise exceptions.StackNotFoundError(
                f"Stack {stack_name} config file doest not exists"
            )
        path = self.get_path(stack_name=stack_name)
        path.unlink()

    def __contains__(self, stack: str | StackConfig) -> bool:
        """
        Check if given stack name or StackConfig exists in context

        Args:
            stack (str | StackConfig): Stack to check

        Returns:
            bool: When stack exists or not
        """
        stack_name = stack.name if isinstance(stack, StackConfig) else stack
        stack_name = f"{stack_name}.json"
        return any(stack_name == f.name for f in self._working_dir.iterdir())

    def __iter__(self) -> t.Iterable[StackConfig]:
        """
        Iter over StackConfig's

        Yields:
            Iterator[t.Iterable[StackConfig]]: Stack config iterable
        """
        for file_path in self._working_dir.iterdir():
            yield StackConfig.from_file(file_path=file_path)

    def __len__(self) -> int:
        return len(list(self._working_dir.iterdir()))

__contains__(stack)

Check if given stack name or StackConfig exists in context

Parameters:

Name Type Description Default
stack str | StackConfig

Stack to check

 required

Returns:

Name Type Description
bool bool

When stack exists or not
Source code in odooghost/context.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def __contains__(self, stack: str | StackConfig) -> bool:
    """
    Check if given stack name or StackConfig exists in context

    Args:
        stack (str | StackConfig): Stack to check

    Returns:
        bool: When stack exists or not
    """
    stack_name = stack.name if isinstance(stack, StackConfig) else stack
    stack_name = f"{stack_name}.json"
    return any(stack_name == f.name for f in self._working_dir.iterdir())

__iter__()

Iter over StackConfig's

Yields:

Type Description
Iterable[StackConfig]

Iterator[t.Iterable[StackConfig]]: Stack config iterable
Source code in odooghost/context.py 
107
108
109
110
111
112
113
114
115
def __iter__(self) -> t.Iterable[StackConfig]:
    """
    Iter over StackConfig's

    Yields:
        Iterator[t.Iterable[StackConfig]]: Stack config iterable
    """
    for file_path in self._working_dir.iterdir():
        yield StackConfig.from_file(file_path=file_path)

create(config)

Create StackConfig file in context

Parameters:

Name Type Description Default
config StackConfig

Stack config

 required

Raises:

Type Description
StackAlreadyExistsError

When stack config file exists
Source code in odooghost/context.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def create(self, config: StackConfig) -> None:
    """
    Create StackConfig file in context

    Args:
        config (StackConfig): Stack config

    Raises:
        exceptions.StackAlreadyExistsError: When stack config file exists
    """
    if config in self:
        raise exceptions.StackAlreadyExistsError(
            f"Stack {config.name} already exists"
        )
    self._write(config=config)

drop(stack_name)

Drop Stack from context

Parameters:

Name Type Description Default
stack_name str

name of stack

 required
Source code in odooghost/context.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
def drop(self, stack_name: str) -> None:
    """
    Drop Stack from context

    Args:
        stack_name (str): name of stack
    """
    if stack_name not in self:
        raise exceptions.StackNotFoundError(
            f"Stack {stack_name} config file doest not exists"
        )
    path = self.get_path(stack_name=stack_name)
    path.unlink()

get(stack_name)

Get StackConfig

Parameters:

Name Type Description Default
stack_name str

name of stack

 required

Returns:

Name Type Description
StackConfig StackConfig

Stack config instance
Source code in odooghost/context.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def get(self, stack_name: str) -> StackConfig:
    """
    Get StackConfig

    Args:
        stack_name (str): name of stack

    Returns:
        StackConfig: Stack config instance
    """
    if stack_name not in self:
        raise exceptions.StackNotFoundError(
            f"Stack {stack_name} config file doest not exists"
        )
    return StackConfig.from_file(file_path=self.get_path(stack_name=stack_name))

get_path(stack_name)

Get Stack config path

Parameters:

Name Type Description Default
stack_name str

name of stack

 required

Returns:

Name Type Description
Path Path

Stack config path
Source code in odooghost/context.py 
21
22
23
24
25
26
27
28
29
30
31
def get_path(self, stack_name: str) -> Path:
    """
    Get Stack config path

    Args:
        stack_name (str): name of stack

    Returns:
        Path: Stack config path
    """
    return self._working_dir / f"{stack_name}.json"

update(config)

Update StackConfig file in context

Parameters:

Name Type Description Default
config StackConfig

Stack config

 required

Raises:

Type Description
StackNotFoundError

When stack config file does not exists
Source code in odooghost/context.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
def update(self, config: StackConfig) -> None:
    """
    Update StackConfig file in context

    Args:
        config (StackConfig): Stack config

    Raises:
        exceptions.StackNotFoundError: When stack config file does not exists
    """
    if config not in self:
        raise exceptions.StackNotFoundError(f"Stack {config.name} not found")
    self._write(config=config)

odooghost.renderer

render_dockerfile(**kw)

Render custom dockerfile for Odoo image

Returns:

Name Type Description
str str

Rendered dockerfile
Source code in odooghost/renderer.py 
14
15
16
17
18
19
20
21
def render_dockerfile(**kw) -> str:
    """
    Render custom dockerfile for Odoo image

    Returns:
        str: Rendered dockerfile
    """
    return env.get_template("Dockerfile.j2").render(**kw)

odooghost.logger

InterceptHandler

Bases: Handler

InterceptHandler convert default logging LogRecord to loguru format

Source code in odooghost/logger.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
class InterceptHandler(logging.Handler):
    """
    InterceptHandler convert default logging LogRecord to loguru format
    """

    def emit(self, record: logging.LogRecord) -> None:
        # Get corresponding Loguru level if it exists
        try:
            level = logger.level(record.levelname).name
        except ValueError:
            level = record.levelno

        # Find caller from where originated the logged message
        frame, depth = logging.currentframe(), 2
        while frame.f_code.co_filename == logging.__file__:
            frame = frame.f_back
            depth += 1

        logger.opt(depth=depth, exception=record.exc_info).log(
            level, record.getMessage()
        )

setup_cli_logging()

Setup logger for CLI

Source code in odooghost/logger.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def setup_cli_logging() -> None:
    """
    Setup logger for CLI
    """
    logger.configure(
        handlers=[
            dict(
                sink=sys.stderr,
                backtrace=True,
                diagnose=True,
                level="DEBUG",
                format="<level>{message}</level>",
            )
        ]
    )