Coverage for slidge/db/alembic/env.py: 62%

1from alembic import context 

2 

3from slidge import global_config 

4from slidge.db.meta import Base, get_engine 

5 

6config = context.config 

7 

8target_metadata = Base.metadata 

9 

10 

11def run_migrations_offline() -> None: 

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

13 

14 This configures the context with just a URL 

15 and not an Engine, though an Engine is acceptable 

16 here as well. By skipping the Engine creation 

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

18 

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

20 script output. 

21 

22 """ 

23 url = config.get_main_option("sqlalchemy.url") 

24 context.configure( 

25 url=url, 

26 target_metadata=target_metadata, 

27 literal_binds=True, 

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

29 render_as_batch=True, 

30 ) 

31 

32 with context.begin_transaction(): 

33 context.run_migrations() 

34 

35 

36def run_migrations_online() -> None: 

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

38 

39 In this scenario we need to create an Engine 

40 and associate a connection with the context. 

41 

42 """ 

43 # for pytest-alembic 

44 connectable = context.config.attributes.get("connection", None) 

45 

46 if connectable is None: 

47 try: 

48 # in prod 

49 connectable = get_engine(global_config.DB_URL) 

50 except AttributeError: 

51 # to generate migrations 

52 connectable = get_engine("sqlite+pysqlite:///dev/slidge.sqlite") 

53 

54 with connectable.connect() as connection: 

55 context.configure( 

56 connection=connection, 

57 target_metadata=target_metadata, 

58 render_as_batch=True, 

59 ) 

60 

61 with context.begin_transaction(): 

62 context.run_migrations() 

63 

64 

65if context.is_offline_mode(): 

66 run_migrations_offline() 

67else: 

68 run_migrations_online()