python_arango_ogm.cli.pao_migrate 

 1import json
 2import os
 3import site
 4import sys
 5
 6import typer
 7from dotenv import load_dotenv
 8
 9from python_arango_ogm.db.pao_database import PAODatabase
10from python_arango_ogm.db.pao_migration_builder import PAOMigrationBuilder
11from python_arango_ogm.db.pao_migrator import PAOMigrator
12
13# print("Setting app root to ", os.getcwd())
14app = typer.Typer()
15
16
17@app.command()
18def make_migrations(env_file: str = None):
19    """
20    Make migrations from models
21    """
22    app_root = get_app_root()
23    load_environment(env_file=env_file)
24    migration_builder = PAOMigrationBuilder(app_root)
25    migration_builder.create_model_migrations()
26
27
28@app.command()
29def new_migration(name: str, env_file: str = None):
30    """
31    Make a new migration with a new number prefix and suffix of given name
32
33    @params:
34        name is the suffix of the migration
35    """
36    app_root = get_app_root()
37    load_environment(env_file=env_file)
38    migration_builder = PAOMigrationBuilder(app_root)
39    migration_builder.create_blank_migration(name)
40
41
42@app.command()
43def migrate(clean: bool = False, env_file: str = None):
44    """ Runs all unapplied migrations.  If clean is true, database will be deleted first. """
45    app_root = get_app_root()
46    load_environment(env_file=env_file)
47    pao_db = PAODatabase(delete_db=clean)
48    migrator = PAOMigrator(pao_db, app_root)
49    migrator.apply_migrations()
50
51
52@app.command()
53def migrate_rollback(env_file: str = None):
54    """ Runs all unapplied migrations """
55    app_root = get_app_root()
56    load_environment(env_file=env_file)
57    pao_db = PAODatabase()
58    migrator = PAOMigrator(pao_db, app_root)
59    migrator.migrate_down()
60
61
62@app.command()
63def list_migrations(env_file: str = None):
64    """ Runs all unapplied migrations """
65    app_root = get_app_root()
66    load_environment(env_file=env_file)
67    pao_db = PAODatabase()
68    migrator = PAOMigrator(pao_db, app_root)
69    migrations = migrator.list_migrations()
70    print(json.dumps(migrations, indent=4))
71
72
73def load_environment(env_file):
74    """ Load given environment file, if it exists. """
75    if env_file:
76        load_dotenv(env_file)
77
78def get_app_root():
79    app_root = os.getcwd()
80    if app_root not in sys.path:
81        site.addsitedir(app_root)
82    return app_root
83
84def run():
85    app()
86
87
88if __name__ == "__main__":
89    run()
app = <typer.main.Typer object>
@app.command()
def make_migrations(env_file: str = None): 
18@app.command()
19def make_migrations(env_file: str = None):
20    """
21    Make migrations from models
22    """
23    app_root = get_app_root()
24    load_environment(env_file=env_file)
25    migration_builder = PAOMigrationBuilder(app_root)
26    migration_builder.create_model_migrations()

Make migrations from models

@app.command()
def new_migration(name: str, env_file: str = None): 
29@app.command()
30def new_migration(name: str, env_file: str = None):
31    """
32    Make a new migration with a new number prefix and suffix of given name
33
34    @params:
35        name is the suffix of the migration
36    """
37    app_root = get_app_root()
38    load_environment(env_file=env_file)
39    migration_builder = PAOMigrationBuilder(app_root)
40    migration_builder.create_blank_migration(name)

Make a new migration with a new number prefix and suffix of given name

@params: name is the suffix of the migration

@app.command()
def migrate(clean: bool = False, env_file: str = None): 
43@app.command()
44def migrate(clean: bool = False, env_file: str = None):
45    """ Runs all unapplied migrations.  If clean is true, database will be deleted first. """
46    app_root = get_app_root()
47    load_environment(env_file=env_file)
48    pao_db = PAODatabase(delete_db=clean)
49    migrator = PAOMigrator(pao_db, app_root)
50    migrator.apply_migrations()

Runs all unapplied migrations. If clean is true, database will be deleted first.

@app.command()
def migrate_rollback(env_file: str = None): 
53@app.command()
54def migrate_rollback(env_file: str = None):
55    """ Runs all unapplied migrations """
56    app_root = get_app_root()
57    load_environment(env_file=env_file)
58    pao_db = PAODatabase()
59    migrator = PAOMigrator(pao_db, app_root)
60    migrator.migrate_down()

Runs all unapplied migrations

@app.command()
def list_migrations(env_file: str = None): 
63@app.command()
64def list_migrations(env_file: str = None):
65    """ Runs all unapplied migrations """
66    app_root = get_app_root()
67    load_environment(env_file=env_file)
68    pao_db = PAODatabase()
69    migrator = PAOMigrator(pao_db, app_root)
70    migrations = migrator.list_migrations()
71    print(json.dumps(migrations, indent=4))

Runs all unapplied migrations

def load_environment(env_file): 
74def load_environment(env_file):
75    """ Load given environment file, if it exists. """
76    if env_file:
77        load_dotenv(env_file)

Load given environment file, if it exists.

def get_app_root(): 
79def get_app_root():
80    app_root = os.getcwd()
81    if app_root not in sys.path:
82        site.addsitedir(app_root)
83    return app_root
def run(): 
85def run():
86    app()