Coverage for /home/martinb/.local/share/virtualenvs/camcops/lib/python3.6/site-packages/cardinal_pythonlib/sqlalchemy/session.py : 67%

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
4"""
5===============================================================================
7 Original code copyright (C) 2009-2021 Rudolf Cardinal (rudolf@pobox.com).
9 This file is part of cardinal_pythonlib.
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
15 https://www.apache.org/licenses/LICENSE-2.0
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.
23===============================================================================
25**Functions to work with SQLAlchemy sessions/engines.**
27"""
29from typing import TYPE_CHECKING
30import os
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
37from cardinal_pythonlib.logs import get_brace_style_log_with_null_handler
39if TYPE_CHECKING:
40 from sqlalchemy.engine.url import URL
42log = get_brace_style_log_with_null_handler(__name__)
45# =============================================================================
46# Database URLs
47# =============================================================================
49SQLITE_MEMORY_URL = "sqlite://"
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 )
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
80# =============================================================================
81# Connection management
82# =============================================================================
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
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)
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))
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)