mirror of
https://github.com/283375/arcaea-offline-pyside-ui.git
synced 2025-11-07 13:02:17 +00:00
refactor: database initialize checker
This commit is contained in:
10
core/database/__init__.py
Normal file
10
core/database/__init__.py
Normal file
@ -0,0 +1,10 @@
|
||||
from .init_checker import DatabaseInitCheckResult, check_db_init
|
||||
from .utils import create_engine, db_path_to_sqlite_url, sqlite_url_to_db_path
|
||||
|
||||
__all__ = [
|
||||
"check_db_init",
|
||||
"create_engine",
|
||||
"db_path_to_sqlite_url",
|
||||
"DatabaseInitCheckResult",
|
||||
"sqlite_url_to_db_path",
|
||||
]
|
||||
30
core/database/init_checker.py
Normal file
30
core/database/init_checker.py
Normal file
@ -0,0 +1,30 @@
|
||||
from enum import Flag, auto
|
||||
from pathlib import Path
|
||||
|
||||
from arcaea_offline.database import Database
|
||||
|
||||
from .utils import create_engine, db_path_to_sqlite_url
|
||||
|
||||
|
||||
class DatabaseInitCheckResult(Flag):
|
||||
NONE = 0
|
||||
FILE_EXISTS = auto()
|
||||
INITIALIZED = auto()
|
||||
|
||||
OK = FILE_EXISTS | INITIALIZED
|
||||
|
||||
|
||||
def check_db_init(file: Path) -> DatabaseInitCheckResult:
|
||||
flags = DatabaseInitCheckResult.NONE
|
||||
|
||||
if not file.exists():
|
||||
return flags
|
||||
|
||||
flags |= DatabaseInitCheckResult.FILE_EXISTS
|
||||
|
||||
db_url = db_path_to_sqlite_url(file)
|
||||
db = Database(create_engine(db_url))
|
||||
if db.check_init():
|
||||
flags |= DatabaseInitCheckResult.INITIALIZED
|
||||
|
||||
return flags
|
||||
28
core/database/utils.py
Normal file
28
core/database/utils.py
Normal file
@ -0,0 +1,28 @@
|
||||
from pathlib import Path
|
||||
|
||||
from PySide6.QtCore import QSysInfo, QUrl
|
||||
from sqlalchemy import Engine
|
||||
from sqlalchemy import create_engine as sa_create_engine
|
||||
from sqlalchemy.pool import NullPool, Pool
|
||||
|
||||
|
||||
def db_path_to_sqlite_url(file: Path) -> QUrl:
|
||||
kernelType = QSysInfo.kernelType()
|
||||
|
||||
# the slash count varies depending on the kernel
|
||||
# https://docs.sqlalchemy.org/en/20/core/engines.html#sqlite
|
||||
uri = file.resolve().as_uri()
|
||||
if kernelType == "winnt":
|
||||
return QUrl(uri.replace("file://", "sqlite://"))
|
||||
else:
|
||||
return QUrl(uri.replace("file://", "sqlite:///"))
|
||||
|
||||
|
||||
def sqlite_url_to_db_path(url: str) -> Path:
|
||||
db_file_url = url.replace("sqlite://", "file://")
|
||||
return Path(QUrl(db_file_url).toLocalFile()).resolve()
|
||||
|
||||
|
||||
def create_engine(_url: str | QUrl, pool: type[Pool] = NullPool) -> Engine:
|
||||
url = _url.toString() if isinstance(_url, QUrl) else _url
|
||||
return sa_create_engine(url, poolclass=pool)
|
||||
Reference in New Issue
Block a user