Hide keyboard shortcuts

Hot-keys on this page

r m x p   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

1#!/usr/bin/env python 

2# cardinal_pythonlib/sqlalchemy/session.py 

3 

4""" 

5=============================================================================== 

6 

7 Original code copyright (C) 2009-2021 Rudolf Cardinal (rudolf@pobox.com). 

8 

9 This file is part of cardinal_pythonlib. 

10 

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

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

13 You may obtain a copy of the License at 

14 

15 https://www.apache.org/licenses/LICENSE-2.0 

16 

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

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

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

20 See the License for the specific language governing permissions and 

21 limitations under the License. 

22 

23=============================================================================== 

24 

25**Functions to work with SQLAlchemy sessions/engines.** 

26 

27""" 

28 

29from typing import TYPE_CHECKING 

30import os 

31 

32from sqlalchemy.engine import create_engine 

33from sqlalchemy.engine.base import Engine 

34from sqlalchemy.engine.url import make_url 

35from sqlalchemy.orm.session import Session 

36 

37from cardinal_pythonlib.logs import get_brace_style_log_with_null_handler 

38 

39if TYPE_CHECKING: 

40 from sqlalchemy.engine.url import URL 

41 

42log = get_brace_style_log_with_null_handler(__name__) 

43 

44 

45# ============================================================================= 

46# Database URLs 

47# ============================================================================= 

48 

49SQLITE_MEMORY_URL = "sqlite://" 

50 

51 

52def make_mysql_url(username: str, password: str, dbname: str, 

53 driver: str = "mysqldb", host: str = "localhost", 

54 port: int = 3306, charset: str = "utf8") -> str: 

55 """ 

56 Makes an SQLAlchemy URL for a MySQL database. 

57 """ 

58 return "mysql+{driver}://{u}:{p}@{host}:{port}/{db}?charset={cs}".format( 

59 driver=driver, 

60 host=host, 

61 port=port, 

62 db=dbname, 

63 u=username, 

64 p=password, 

65 cs=charset, 

66 ) 

67 

68 

69def make_sqlite_url(filename: str) -> str: 

70 """ 

71 Makes an SQLAlchemy URL for a SQLite database. 

72 """ 

73 absfile = os.path.abspath(filename) 

74 return "sqlite://{host}/{path}".format(host="", path=absfile) 

75 # ... makes it clear how it works! Ends up being sqlite:////abspath 

76 # or sqlite:///relpath. Also works with backslashes for Windows paths; see 

77 # http://docs.sqlalchemy.org/en/latest/core/engines.html#sqlite 

78 

79 

80# ============================================================================= 

81# Connection management 

82# ============================================================================= 

83 

84def get_engine_from_session(dbsession: Session) -> Engine: 

85 """ 

86 Gets the SQLAlchemy :class:`Engine` from a SQLAlchemy :class:`Session`. 

87 """ 

88 engine = dbsession.bind 

89 assert isinstance(engine, Engine) 

90 return engine 

91 

92 

93def get_safe_url_from_engine(engine: Engine) -> str: 

94 """ 

95 Gets a URL from an :class:`Engine`, obscuring the password. 

96 """ 

97 raw_url = engine.url # type: str 

98 url_obj = make_url(raw_url) # type: URL 

99 return repr(url_obj) 

100 # The default repr() implementation calls 

101 # self.__to_string__(hide_password=False) 

102 

103 

104def get_safe_url_from_session(dbsession: Session) -> str: 

105 """ 

106 Gets a URL from a :class:`Session`, obscuring the password. 

107 """ 

108 return get_safe_url_from_engine(get_engine_from_session(dbsession)) 

109 

110 

111def get_safe_url_from_url(url: str) -> str: 

112 """ 

113 Converts an SQLAlchemy URL into a safe version that obscures the password. 

114 """ 

115 engine = create_engine(url) 

116 return get_safe_url_from_engine(engine)