Coverage for app/alembic/env.py: 0%

1from logging.config import fileConfig 

2 

3from alembic import context 

4from sqlalchemy import engine_from_config, pool 

5 

6# this is the Alembic Config object, which provides 

7# access to the values within the .ini file in use. 

8config = context.config 

9 

10# Interpret the config file for Python logging. 

11# This line sets up loggers basically. 

12if config.config_file_name is not None: 

13 fileConfig(config.config_file_name) 

14 

15# add your model's MetaData object here 

16# for 'autogenerate' support 

17# from myapp import mymodel 

18# target_metadata = mymodel.Base.metadata 

19target_metadata = None 

20 

21from app.core.config import settings # noqa 

22 

23# other values from the config, defined by the needs of env.py, 

24# can be acquired: 

25# my_important_option = config.get_main_option("my_important_option") 

26# ... etc. 

27 

28 

29def get_url() -> str: 

30 return str(settings.SQLALCHEMY_DATABASE_URI) 

31 

32 

33def run_migrations_offline() -> None: 

34 """Run migrations in 'offline' mode. 

35 

36 This configures the context with just a URL 

37 and not an Engine, though an Engine is acceptable 

38 here as well. By skipping the Engine creation 

39 we don't even need a DBAPI to be available. 

40 

41 Calls to context.execute() here emit the given string to the 

42 script output. 

43 

44 """ 

45 url = get_url() 

46 context.configure( 

47 url=url, 

48 target_metadata=target_metadata, 

49 literal_binds=True, 

50 dialect_opts={"paramstyle": "named"}, 

51 ) 

52 

53 with context.begin_transaction(): 

54 context.run_migrations() 

55 

56 

57def run_migrations_online() -> None: 

58 """Run migrations in 'online' mode. 

59 

60 In this scenario we need to create an Engine 

61 and associate a connection with the context. 

62 

63 """ 

64 configuration = config.get_section(config.config_ini_section) 

65 configuration["sqlalchemy.url"] = get_url() 

66 connectable = engine_from_config( 

67 configuration, 

68 prefix="sqlalchemy.", 

69 poolclass=pool.NullPool, 

70 ) 

71 

72 with connectable.connect() as connection: 

73 context.configure(connection=connection, target_metadata=target_metadata) 

74 

75 with context.begin_transaction(): 

76 context.run_migrations() 

77 

78 

79if context.is_offline_mode(): 

80 run_migrations_offline() 

81else: 

82 run_migrations_online()