Last active
March 28, 2020 21:27
-
-
Save compwron/6a237a88cc2b4ed463e923d01859b521 to your computer and use it in GitHub Desktop.
demo of how I make sqlite pretend to be mysql better so we can test code which runs raw sql
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# conftest.py | |
import datetime | |
from time import strptime | |
from dateutil import parser | |
from sqlalchemy import create_engine | |
def is_test(): | |
""" | |
Tests which want to run against raw sql can override this to be True | |
Example: | |
@patch('my_app.helpers.raw_sql_runner.is_test', side_effect=lambda: True) | |
def test_run_raw_sql_string(self, is_test_mock, session): | |
""" | |
return False | |
def run_raw_sql(session, sql, params): | |
if is_test(): | |
interpolated_sql = sql | |
for key, val in params.items(): | |
if isinstance(val, (list, tuple)): | |
if isinstance(val[0], str): | |
str_vals = map(lambda x: f"'{x}'", val) | |
interpolated_sql = interpolated_sql.replace(f':{key}', f"({','.join(str_vals)})") | |
else: | |
interpolated_sql = interpolated_sql.replace(f':{key}', f"({','.join(map(str, val))})") | |
elif isinstance(val, str): | |
interpolated_sql = interpolated_sql.replace(f':{key}', f"'{str(val)}'") | |
else: | |
interpolated_sql = interpolated_sql.replace(f':{key}', str(val)) | |
return session.execute(interpolated_sql, params) | |
else: | |
return session.execute(sql, params) | |
def fake_sqlite_datediff(d1, d2): | |
d1 = datetime.strptime(d1, "%Y-%m-%d") | |
d2 = datetime.strptime(d2, "%Y-%m-%d") | |
return abs((d2 - d1).days) | |
def fake_sqlite_convert_tz(date_str, _tz1, _tz2): | |
return date_str | |
def fake_sqlite_hour(_date_str): | |
return 12 | |
def fake_sqlite_yearweek(some_date): | |
return parser.isoparse(some_date).strftime("%Y%V") | |
def fake_sqlite_concat(a, b): | |
return f'{a}{b}' | |
def fake_sqlite_str_to_date(date_str, format): | |
# https://www.w3schools.com/sql/func_mysql_str_to_date.asp | |
# https://docs.python.org/3/library/datetime.html | |
mysql_to_python_format_mapping = { | |
'X': 'Y', # year | |
'V': 'U', # week of year | |
' ': '', # what the heck mysql | |
'W': 'A', # long name of week | |
} | |
for key, value in mysql_to_python_format_mapping.items(): | |
format = format.replace(key, value) | |
return strptime(date_str, format) | |
def fake_sqlite_now(): | |
return '2019-01-01 00:00:00' | |
def fake_sqlite_subdate(_date_str, _interval): | |
return '2019-01-01 00:00:00' | |
def create_sqlite_functions(con): | |
con.create_function("datediff", 2, fake_sqlite_datediff) | |
con.create_function("convert_tz", 3, fake_sqlite_convert_tz) | |
con.create_function("hour", 1, fake_sqlite_hour) | |
con.create_function("yearweek", 1, fake_sqlite_yearweek) | |
con.create_function("concat", 2, fake_sqlite_concat) | |
con.create_function("str_to_date", 2, fake_sqlite_str_to_date) | |
con.create_function("now", 0, fake_sqlite_now) | |
con.create_function("subdate", 2, fake_sqlite_subdate) | |
def get_database_settings(): | |
# if stuff, real setup, else, sqlite | |
return {'name_or_url': 'sqlite://'} | |
engine_settings = get_database_settings() | |
engine = create_engine(**engine_settings) | |
con = engine.connect().connection | |
create_sqlite_functions(con) | |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment