| from __future__ import annotations | |
| import time | |
| from sqlalchemy import text, func | |
| _DB_TYPE: str = "postgresql" | |
| def set_db_type(db_url: str): | |
| global _DB_TYPE | |
| _DB_TYPE = "sqlite" if db_url.startswith("sqlite") else "postgresql" | |
| def is_sqlite() -> bool: | |
| return _DB_TYPE == "sqlite" | |
| def now_expr(): | |
| return text("CURRENT_TIMESTAMP") | |
| def timestamp_ge_expr(column, ts: float): | |
| if is_sqlite(): | |
| return column >= func.datetime(ts, "unixepoch") | |
| return column >= func.to_timestamp(ts) | |
| def timestamp_lt_expr(column, ts: float): | |
| if is_sqlite(): | |
| return func.datetime(column) < func.datetime(ts, "unixepoch") | |
| return column < func.to_timestamp(ts) | |
| def strftime_expr(format_str: str, column): | |
| if is_sqlite(): | |
| return func.strftime(format_str, column) | |
| return func.to_char(column, format_str) | |