c8faab97d9
- Introduced a helper function `_to_sync_database_url` to standardize the conversion of async database URLs to sync URLs in `env.py` and `migrations.py`. - Added `_ensure_sqlite_parent_dir` function in `db_utils.py` to create the parent directory for SQLite databases if it doesn't exist, ensuring compatibility with Windows paths. - Updated the `get_database_url_and_connect_args` function to call `_ensure_sqlite_parent_dir` before processing the database URL. - Simplified the `sync_export_runtime.js` script by removing the build step and directly using committed runtime artifacts, ensuring a smoother export process.
68 lines
2.3 KiB
Python
68 lines
2.3 KiB
Python
import os
|
|
from utils.get_env import get_app_data_directory_env, get_database_url_env
|
|
from urllib.parse import urlsplit, urlunsplit, parse_qsl
|
|
import ssl
|
|
|
|
|
|
def _ensure_sqlite_parent_dir(database_url: str) -> None:
|
|
if not database_url.startswith("sqlite://"):
|
|
return
|
|
|
|
split_result = urlsplit(database_url)
|
|
db_path = split_result.path
|
|
if not db_path:
|
|
return
|
|
|
|
# sqlite URLs on Windows can start with /C:/..., normalize that for os.path.
|
|
if os.name == "nt" and len(db_path) >= 3 and db_path[0] == "/" and db_path[2] == ":":
|
|
db_path = db_path[1:]
|
|
|
|
parent = os.path.dirname(db_path)
|
|
if parent:
|
|
os.makedirs(parent, exist_ok=True)
|
|
|
|
|
|
def get_database_url_and_connect_args() -> tuple[str, dict]:
|
|
database_url = get_database_url_env() or "sqlite:///" + os.path.join(
|
|
get_app_data_directory_env() or "/tmp/presenton", "fastapi.db"
|
|
)
|
|
|
|
_ensure_sqlite_parent_dir(database_url)
|
|
|
|
if database_url.startswith("sqlite://"):
|
|
database_url = database_url.replace("sqlite://", "sqlite+aiosqlite://", 1)
|
|
elif database_url.startswith("postgresql://"):
|
|
database_url = database_url.replace("postgresql://", "postgresql+asyncpg://", 1)
|
|
elif database_url.startswith("mysql://"):
|
|
database_url = database_url.replace("mysql://", "mysql+aiomysql://", 1)
|
|
else:
|
|
database_url = database_url
|
|
|
|
connect_args = {}
|
|
if "sqlite" in database_url:
|
|
connect_args["check_same_thread"] = False
|
|
|
|
try:
|
|
split_result = urlsplit(database_url)
|
|
if split_result.query:
|
|
query_params = parse_qsl(split_result.query, keep_blank_values=True)
|
|
driver_scheme = split_result.scheme
|
|
for k, v in query_params:
|
|
key_lower = k.lower()
|
|
if key_lower == "sslmode" and "postgresql+asyncpg" in driver_scheme:
|
|
if v.lower() != "disable" and "sqlite" not in database_url:
|
|
connect_args["ssl"] = ssl.create_default_context()
|
|
|
|
database_url = urlunsplit(
|
|
(
|
|
split_result.scheme,
|
|
split_result.netloc,
|
|
split_result.path,
|
|
"",
|
|
split_result.fragment,
|
|
)
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
return database_url, connect_args
|