u

core/database/link.py

@@ -0,0 +1,18 @@
+import os
+
+from core.config import Config
+from core.constants import db_path_default, database_path
+
+db_link = Config("db_path", default=db_path_default, secret=True)
+db_path = database_path
+
+db_type = db_link.split("://")[0].split("+")[0]
+if db_type == "sqlite":
+    db_path = os.path.dirname(db_link.replace("sqlite:///", ""))
+os.makedirs(db_path, exist_ok=True)
+
+def get_db_link(orm_type="sqlalchemy"):
+    if orm_type == "sqlalchemy":
+        return db_link
+    if orm_type == "tortoise":
+        return db_type + "://" + db_link.split("://")[1]

core/database/orm.py

@@ -1,26 +1,15 @@
-import os
-
 from sqlalchemy import create_engine
 from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
 from sqlalchemy.orm import sessionmaker
 
-from core.config import Config
-from core.constants.default import db_path_default
-from core.constants.path import database_path
+from core.database.link import db_link
 from core.database.orm_base import Base
 
-DB_LINK = Config("db_path", default=db_path_default, secret=True)
-
-db_path = database_path
-if DB_LINK.startswith("sqlite:///"):
-    db_path = os.path.dirname(DB_LINK.replace("sqlite:///", ""))
-os.makedirs(db_path, exist_ok=True)
-
 
 class DBSession:
     def __init__(self):
         self.engine = create_engine(
-            DB_LINK, isolation_level="READ UNCOMMITTED", pool_pre_ping=True
+            db_link, isolation_level="READ UNCOMMITTED", pool_pre_ping=True
         )
         self.Session = sessionmaker()
         self.Session.configure(bind=self.engine)
@@ -35,7 +24,7 @@ def create(self):
 
 class AsyncDBSession:
     def __init__(self):
-        self.engine = create_async_engine(DB_LINK, isolation_level="READ UNCOMMITTED")
+        self.engine = create_async_engine(db_link, isolation_level="READ UNCOMMITTED")
         self.Session = async_sessionmaker()
         self.Session.configure(bind=self.engine)
 

core/database/tables.py

@@ -2,10 +2,11 @@
 from sqlalchemy.dialects.mysql import LONGTEXT
 
 from core.config import Config
-from core.database.orm import Session, DB_LINK
+from core.database.orm import Session
+from core.database.link import db_type
 from core.database.orm_base import Base
 
-is_mysql = DB_LINK.startswith("mysql")
+is_mysql = db_type == "mysql"
 default_locale = Config("default_locale", cfg_type=str)