Coverage for adhoc-cicd-odoo-odoo / odoo / modules / migration.py: 27%
117 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-09 18:05 +0000
1# Part of Odoo. See LICENSE file for full copyright and licensing details.
3""" Modules migration handling. """
4from __future__ import annotations
6import glob
7import inspect
8import itertools
9import logging
10import os
11import re
12import typing
13from collections import defaultdict
14from os.path import join as opj
16import odoo.release as release
17import odoo.upgrade
18from odoo.modules.module import load_script
19from odoo.orm.registry import Registry
20from odoo.tools.misc import file_path
21from odoo.tools.parse_version import parse_version
23if typing.TYPE_CHECKING:
24 from collections.abc import Iterator
25 from odoo.sql_db import Cursor
26 from . import module_graph
28_logger = logging.getLogger(__name__)
31VERSION_RE = re.compile(
32 r"""^
33 # Optional prefix with Odoo version
34 ((
35 6\.1|
37 # "x.0" version, with x >= 6.
38 [6-9]\.0|
40 # multi digits "x.0" versions
41 [1-9]\d+\.0|
43 # x.saas~y, where x >= 7 and x <= 10
44 (7|8|9|10)\.saas~[1-9]\d*|
46 # saas~x.y, where x >= 11 and y between 1 and 9
47 # FIXME handle version >= saas~100 (expected in year 2106)
48 saas~(1[1-9]|[2-9]\d+)\.[1-9]
49 )\.)?
50 # After Odoo version we allow precisely 2 or 3 parts
51 # note this will also allow 0.0.0 which has a special meaning
52 \d+\.\d+(\.\d+)?
53 $""",
54 re.VERBOSE | re.ASCII,
55)
58class MigrationManager:
59 """ Manages the migration of modules.
61 Migrations files must be python files containing a ``migrate(cr, installed_version)``
62 function. These files must respect a directory tree structure: A 'migrations' folder
63 which contains a folder by version. Version can be 'module' version or 'server.module'
64 version (in this case, the files will only be processed by this version of the server).
65 Python file names must start by ``pre-`` or ``post-`` and will be executed, respectively,
66 before and after the module initialisation. ``end-`` scripts are run after all modules have
67 been updated.
69 A special folder named ``0.0.0`` can contain scripts that will be run on any version change.
70 In `pre` stage, ``0.0.0`` scripts are run first, while in ``post`` and ``end``, they are run last.
72 Example::
74 <moduledir>
75 `-- migrations
76 |-- 1.0
77 | |-- pre-update_table_x.py
78 | |-- pre-update_table_y.py
79 | |-- post-create_plop_records.py
80 | |-- end-cleanup.py
81 | `-- README.txt # not processed
82 |-- 9.0.1.1 # processed only on a 9.0 server
83 | |-- pre-delete_table_z.py
84 | `-- post-clean-data.py
85 |-- 0.0.0
86 | `-- end-invariants.py # processed on all version update
87 `-- foo.py # not processed
88 """
89 migrations: defaultdict[str, dict]
91 def __init__(self, cr: Cursor, graph: module_graph.ModuleGraph):
92 self.cr = cr
93 self.graph = graph
94 self.migrations = defaultdict(dict)
95 self._get_files()
97 def _get_files(self) -> None:
98 def _get_upgrade_path(pkg: str) -> Iterator[str]:
99 for path in odoo.upgrade.__path__:
100 upgrade_path = opj(path, pkg)
101 if os.path.exists(upgrade_path):
102 yield upgrade_path
104 def _verify_upgrade_version(path: str, version: str) -> bool:
105 full_path = opj(path, version)
106 if not os.path.isdir(full_path):
107 return False
109 if version == "tests":
110 return False
112 if not VERSION_RE.match(version):
113 _logger.warning("Invalid version for upgrade script %r", full_path)
114 return False
116 return True
118 def get_scripts(path: str) -> dict[str, list[str]]:
119 if not path:
120 return {}
121 return {
122 version: glob.glob(opj(path, version, '*.py'))
123 for version in os.listdir(path)
124 if _verify_upgrade_version(path, version)
125 }
127 def check_path(path: str) -> str:
128 try:
129 return file_path(path)
130 except FileNotFoundError:
131 return ''
133 for pkg in self.graph:
134 if pkg.load_state != 'to upgrade' and pkg.name not in Registry(self.cr.dbname)._force_upgrade_scripts: 134 ↛ 138line 134 didn't jump to line 138 because the condition on line 134 was always true
135 continue
138 self.migrations[pkg.name] = {
139 'module': get_scripts(check_path(pkg.name + '/migrations')),
140 'module_upgrades': get_scripts(check_path(pkg.name + '/upgrades')),
141 }
143 scripts = defaultdict(list)
144 for p in _get_upgrade_path(pkg.name):
145 for v, s in get_scripts(p).items():
146 scripts[v].extend(s)
147 self.migrations[pkg.name]["upgrade"] = scripts
149 def migrate_module(self, pkg: module_graph.ModuleNode, stage: typing.Literal['pre', 'post', 'end']) -> None:
150 assert stage in ('pre', 'post', 'end')
151 stageformat = {
152 'pre': '[>%s]',
153 'post': '[%s>]',
154 'end': '[$%s]',
155 }
156 if pkg.load_state != 'to upgrade' and pkg.name not in Registry(self.cr.dbname)._force_upgrade_scripts: 156 ↛ 159line 156 didn't jump to line 159 because the condition on line 156 was always true
157 return
159 def convert_version(version: str) -> str:
160 if version == "0.0.0":
161 return version
162 if version.count(".") > 2:
163 return version # the version number already contains the server version, see VERSION_RE for details
164 return "%s.%s" % (release.major_version, version)
166 def _get_migration_versions(pkg, stage: str) -> list[str]:
167 versions = sorted({
168 ver: None
169 for lv in self.migrations[pkg.name].values()
170 for ver, lf in lv.items()
171 if lf
172 }, key=lambda k: parse_version(convert_version(k)))
173 if "0.0.0" in versions:
174 # reorder versions
175 versions.remove("0.0.0")
176 if stage == "pre":
177 versions.insert(0, "0.0.0")
178 else:
179 versions.append("0.0.0")
180 return versions
182 def _get_migration_files(pkg, version, stage):
183 """ return a list of migration script files
184 """
185 m = self.migrations[pkg.name]
187 return sorted(
188 (
189 f
190 for k in m
191 for f in m[k].get(version, [])
192 if os.path.basename(f).startswith(f"{stage}-")
193 ),
194 key=os.path.basename,
195 )
197 installed_version = pkg.load_version or ''
198 parsed_installed_version = parse_version(installed_version)
199 current_version = parse_version(convert_version(pkg.manifest['version']))
201 def compare(version: str) -> bool:
202 if version == "0.0.0" and parsed_installed_version < current_version:
203 return True
205 full_version = convert_version(version)
206 majorless_version = (version != full_version)
208 if majorless_version:
209 # We should not re-execute major-less scripts when upgrading to new Odoo version
210 # a module in `9.0.2.0` should not re-execute a `2.0` script when upgrading to `10.0.2.0`.
211 # In which case we must compare just the module version
212 return parsed_installed_version[2:] < parse_version(full_version)[2:] <= current_version[2:]
214 return parsed_installed_version < parse_version(full_version) <= current_version
216 versions = _get_migration_versions(pkg, stage)
217 for version in versions:
218 if compare(version):
219 for pyfile in _get_migration_files(pkg, version, stage):
220 exec_script(self.cr, installed_version, pyfile, pkg.name, stage, stageformat[stage] % version)
223VALID_MIGRATE_PARAMS = list(itertools.product(
224 ['cr', '_cr'],
225 ['version', '_version'],
226))
228def exec_script(cr, installed_version, pyfile, addon, stage, version=None):
229 version = version or installed_version
230 name, ext = os.path.splitext(os.path.basename(pyfile))
231 if ext.lower() != '.py':
232 return
233 try:
234 mod = load_script(pyfile, name)
235 except ImportError as e:
236 raise ImportError('module %(addon)s: Unable to load %(stage)s-migration file %(file)s' % dict(locals(), file=pyfile)) from e
238 if not hasattr(mod, 'migrate'):
239 raise AttributeError(
240 'module %(addon)s: Each %(stage)s-migration file must have a "migrate(cr, installed_version)" function, not found in %(file)s' % dict(
241 locals(),
242 file=pyfile,
243 ))
245 try:
246 sig = inspect.signature(mod.migrate)
247 except TypeError as e:
248 raise TypeError("module %(addon)s: `migrate` needs to be a function, got %(migrate)r" % dict(locals(), migrate=mod.migrate)) from e
250 if not (
251 tuple(sig.parameters.keys()) in VALID_MIGRATE_PARAMS
252 and all(p.kind in (p.POSITIONAL_ONLY, p.POSITIONAL_OR_KEYWORD) for p in sig.parameters.values())
253 ):
254 raise TypeError("module %(addon)s: `migrate`'s signature should be `(cr, version)`, %(func)s is %(sig)s" % dict(locals(), func=mod.migrate, sig=sig))
256 _logger.info('module %(addon)s: Running migration %(version)s %(name)s' % dict(locals(), name=mod.__name__)) # noqa: G002
257 mod.migrate(cr, installed_version)