import os
from pathlib import Path
import sys
if sys.version_info < (3, 10):
from importlib_metadata import entry_points
else:
from importlib.metadata import entry_points
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 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_schema = module_dist.entry_points["config_schema"].load()
config = {"MODULE_CODE": module_code, "MODULE_URL": f"/{module_code.lower()}"}
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 Exception(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 alembic_branch_in_use(branch_name, directory, x_arg):
"""
Return true if at least one revision of the given branch is applied.
"""
db = current_app.extensions["sqlalchemy"].db
migrate = current_app.extensions["migrate"].migrate
config = migrate.get_config(directory, x_arg)
script = ScriptDirectory.from_config(config)
base_revision = script.get_revision(script.as_revision_number(branch_name))
branch_revisions = set(iterate_revisions(script, base_revision.revision))
migration_context = MigrationContext.configure(db.session.connection())
current_heads = migration_context.get_current_heads()
# get_current_heads does not return implicit revision through dependencies, get_all_current does
current_heads = set(map(lambda rev: rev.revision, script.get_all_current(current_heads)))
return not branch_revisions.isdisjoint(current_heads)
[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_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_code.capitalize(),
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