Coverage for /builds/BuildGrid/buildgrid/buildgrid/server/sql/alembic/env.py: 100.00%

1# Copyright (C) 2019 Bloomberg LP 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# <http://www.apache.org/licenses/LICENSE-2.0> 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14 

15from logging.config import fileConfig 

16 

17from alembic import context 

18from sqlalchemy import engine_from_config, pool 

19from sqlalchemy.engine.base import Connection 

20 

21from buildgrid.server.sql.models import Base 

22 

23# this is the Alembic Config object, which provides 

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

25config = context.config 

26 

27# Interpret the config file for Python logging. 

28# This line sets up loggers basically. 

29try: 

30 fileConfig(config.config_file_name) # type: ignore[arg-type] # covariance in union issue. Not a real error. 

31except TypeError: 

32 pass 

33 

34target_metadata = Base.metadata 

35 

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

37# can be acquired: 

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

39# ... etc. 

40 

41 

42def run_migrations_offline() -> None: # pragma: no cover 

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

44 

45 This configures the context with just a URL 

46 and not an Engine, though an Engine is acceptable 

47 here as well. By skipping the Engine creation 

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

49 

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

51 script output. 

52 

53 """ 

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

55 context.configure(url=url, target_metadata=target_metadata, literal_binds=True) 

56 

57 with context.begin_transaction(): 

58 context.run_migrations() 

59 

60 

61def _run_migrations_online(connection: Connection) -> None: # pragma: no cover 

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

63 

64 with context.begin_transaction(): 

65 context.run_migrations() 

66 

67 

68def run_migrations_online() -> None: # pragma: no cover 

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

70 

71 In this scenario we need to create an Engine 

72 and associate a connection with the context. 

73 

74 """ 

75 try: 

76 connection = config.attributes["connection"] 

77 _run_migrations_online(connection) 

78 

79 except KeyError: 

80 connectable = engine_from_config( 

81 config.get_section(config.config_ini_section), prefix="sqlalchemy.", poolclass=pool.NullPool 

82 ) 

83 with connectable.connect() as connection: 

84 _run_migrations_online(connection) 

85 

86 

87if context.is_offline_mode(): # pragma: no cover 

88 run_migrations_offline() 

89else: # pragma: no cover 

90 run_migrations_online()