import os
from pathlib import Path
import sys
if sys.version_info < (3, 10):
from importlib_metadata import entry_points, PackageNotFoundError
else:
from importlib.metadata import entry_points, PackageNotFoundError
from alembic.script import ScriptDirectory
from alembic.migration import MigrationContext
from flask import current_app
from flask_migrate import upgrade as db_upgrade
from geonature.utils.utilstoml import load_and_validate_toml
from geonature.utils.env import db, CONFIG_FILE
from geonature.core.gn_commons.models import TModules
from sqlalchemy import exists, select
[docs]
def iter_modules_dist():
for module_code_entry in set(entry_points(group="gn_module", name="code")):
yield module_code_entry.dist
[docs]
def get_module_config_path(module_code):
config_path = os.environ.get(f"GEONATURE_{module_code}_CONFIG_FILE")
if config_path:
return Path(config_path)
config_path = Path(CONFIG_FILE).parent / f"{module_code.lower()}_config.toml"
if config_path.exists():
return config_path
dist = get_dist_from_code(module_code)
module_path = Path(sys.modules[dist.entry_points["code"].module].__file__).parent
# module_path is commonly backend/gn_module_XXX/ but config dir is at package root
config_path = module_path.parent.parent / "config" / "conf_gn_module.toml"
if config_path.exists():
return config_path
return None
[docs]
def get_module_config(module_dist):
module_code = module_dist.entry_points["code"].load()
config = {
"MODULE_CODE": module_code,
"MODULE_URL": f"/{module_code.lower()}", # path to the module in the frontend
"MODULE_API": f"/{module_code.lower()}", # path to the module API
}
try:
config_schema = module_dist.entry_points["config_schema"].load()
except KeyError:
pass # this module does not have any config
else:
config.update(load_and_validate_toml(get_module_config_path(module_code), config_schema))
return config
[docs]
def get_dist_from_code(module_code):
for dist in iter_modules_dist():
if module_code == dist.entry_points["code"].load():
return dist
raise PackageNotFoundError(f"Module with code {module_code} not installed in venv")
[docs]
def iterate_revisions(script, base_revision):
"""
Iterate revisions without following depends_on directive.
Useful to find all revisions of a given branch.
"""
yelded = set()
todo = {base_revision}
while todo:
rev = todo.pop()
yield rev
yelded.add(rev)
rev = script.get_revision(rev)
todo |= rev.nextrev - yelded
[docs]
def get_script_from_config(directory: str, x_arg: []) -> ScriptDirectory:
migrate = current_app.extensions["migrate"].migrate
config = migrate.get_config(directory, x_arg)
script = ScriptDirectory.from_config(config)
return script
[docs]
def get_all_current_alembic_heads(directory: str, x_arg: []) -> set:
script = get_script_from_config(directory, x_arg)
db = current_app.extensions["sqlalchemy"].db
migration_context = MigrationContext.configure(db.session.connection())
current_heads = migration_context.get_current_heads()
# get_current_heads does not return implicit revision through dependencies,
# while get_all_current does
all_current_heads = set(map(lambda rev: rev.revision, script.get_all_current(current_heads)))
return all_current_heads
[docs]
def get_script_and_base_revision_for_one_branch(branch_name: str, directory, x_arg) -> ():
script = get_script_from_config(directory, x_arg)
base_revision = script.get_revision(script.as_revision_number(branch_name))
return script, base_revision
[docs]
def get_all_revisions_for_one_branch(branch_name: str, directory, x_arg) -> set:
script, base_revision = get_script_and_base_revision_for_one_branch(
branch_name, directory, x_arg
)
branch_revisions = set(iterate_revisions(script, base_revision.revision))
return branch_revisions
[docs]
def get_last_revision_for_one_branch(branch_name: str, directory, x_arg) -> set:
script, base_revision = get_script_and_base_revision_for_one_branch(
branch_name, directory, x_arg
)
*_, last_revision = iterate_revisions(script, base_revision.revision)
return last_revision
[docs]
def alembic_branch_in_use(branch_name: str, directory, x_arg):
"""Is an Alembic branch in use.
Return `True` if at least one revision of the given branch is stamped.
Returns
-------
bool
`True` if the Alembic branch has at least one revision stamped,
`False` otherwise.
"""
branch_revisions = get_all_revisions_for_one_branch(branch_name, directory, x_arg)
all_current_alembic_heads = get_all_current_alembic_heads(directory, x_arg)
return not branch_revisions.isdisjoint(all_current_alembic_heads)
[docs]
def is_alembic_branch_up_to_date(branch_name: str, directory: str = None, x_arg: list = []) -> bool:
"""Is a specific Alembic branch fully stamped.
Parameters
----------
branch_name : str
The name of the Alembic branch.
directory : str
The name of the directory containing the migrations files for the branch.
Value of `None` defaults to "migrations".
x_arg : list
Additional arguments consumed by custom `env.py` scripts.
Returns
-------
bool
`True` if the Alembic branch has all its revisions stamped,
`False` otherwise.
"""
head_revision_of_branch = get_last_revision_for_one_branch(branch_name, directory, x_arg)
all_current_alembic_heads = get_all_current_alembic_heads(directory, x_arg)
return head_revision_of_branch in all_current_alembic_heads
[docs]
def exists_in_t_modules(module_code: str):
"""
Returns True if a module with the given code exists in the database
Parameters
----------
module_code : str
The code of the module (for ex. : SYNTHESE, OCCHAB, etc...)
"""
return db.session.execute(
exists(TModules).where(TModules.module_code == module_code).select()
).scalar()
[docs]
def is_module_installed(
python_module_name: str,
migrations_dir: str = None,
alembic_branch_name: str = None,
check_if_all_revisions_have_been_applied: bool = True,
):
"""Is a GeoNature module installed.
We consider a module is installed if and only if:
- The Python package is installed
and (
`check_if_all_revisions_have_been_applied` == true
and
(
all its Alembic migrations are stamped
or
there is no Alembic migration at all for the module
)
)
and the module is registered in the database
Parameters
----------
python_module_name : str
The name of the python module.
Can be found in the root file "setup.py" of the module repository.
Examples:
- "gn_module_occhab"
- "occtax"
- "gn_module_validation"
- "gn_module_dashboard"
- "gn_module_export"
- "gn_module_monitoring"
migrations_dir : str
The name of the directory containing the migrations files for the branch.
Value of `None` defaults to "migrations".
alembic_branch_name : str
The name of the Alembic branch.
Value of `None` defaults to the module code in lowercase.
check_if_all_revisions_have_been_applied : bool
Check if all revisions of the module have been applied.
Returns
-------
bool
`True` if the module is installed,
`False` otherwise.
"""
try:
# Verify if the module Python package is actually installed
init_module = __import__(python_module_name + ".__init__")
module_code = init_module.MODULE_CODE
except ImportError:
# Module not installed because Python package not installed
return False
# Verify if the module is registered in the database
if not exists_in_t_modules(module_code):
# Module not installed because not registered in the database
return False
if check_if_all_revisions_have_been_applied:
try:
# Verify if there are migrations
str_import_path_migrations_dir = python_module_name
if not migrations_dir:
str_import_path_migrations_dir += ".migrations"
else:
str_import_path_migrations_dir += f".{migrations_dir}"
__import__(str_import_path_migrations_dir)
# If there are migrations, check if the branch is up to date
if not alembic_branch_name:
alembic_branch_name = module_code.lower()
if alembic_branch_name and is_alembic_branch_up_to_date(
alembic_branch_name, migrations_dir
):
# Module installed, case with Alembic migrations
return True
except ImportError:
# Module installed, case without Alembic migrations
return True
# Module not installed because Alembic branch not up to date
return False
return True
[docs]
def module_db_upgrade(module_dist, directory=None, sql=False, tag=None, x_arg=[]):
module_code = module_dist.entry_points["code"].load()
module_blueprint = module_dist.entry_points["blueprint"].load() # force discovery of models
if module_dist.entry_points.select(name="migrations"):
try:
alembic_branch = module_dist.entry_points["alembic_branch"].load()
except KeyError:
alembic_branch = module_code.lower()
else:
alembic_branch = None
module = db.session.execute(
select(TModules).filter_by(module_code=module_code)
).scalar_one_or_none()
if module is None:
# add module to database
try:
module_label = module_dist.entry_points["label"].load()
except KeyError:
module_label = module_code.capitalize()
try:
module_picto = module_dist.entry_points["picto"].load()
except KeyError:
module_picto = "fa-puzzle-piece"
try:
module_type = module_dist.entry_points["type"].load()
except KeyError:
module_type = None
try:
module_doc_url = module_dist.entry_points["doc_url"].load()
except KeyError:
module_doc_url = None
module = TModules(
type=module_type,
module_code=module_code,
module_label=module_label,
module_path=module_code.lower(),
module_target="_self",
module_picto=module_picto,
module_doc_url=module_doc_url,
active_frontend=True,
active_backend=True,
ng_module=module_code.lower(),
)
db.session.add(module)
db.session.commit()
elif alembic_branch and not alembic_branch_in_use(alembic_branch, directory, x_arg):
"""
The module branch is not known to be applied by Alembic,
but the module is present in gn_commons.t_modules table.
Refusing to upgrade the Alembic branch.
Upgrading of old module requiring manual stamp?
"""
return False
if alembic_branch:
revision = alembic_branch + "@head"
db_upgrade(directory, revision, sql, tag, x_arg)
return True
[docs]
def get_module_version(module_code: str):
"""
Get the module version from the module_code. We check what python package is installed.
If no package is found, we return None.
"""
try:
if module_code:
dist = get_dist_from_code(module_code)
return dist.version
except PackageNotFoundError:
return None
return None