fix: move blocking io ops to async method on dbservice init (#5291)
* move blocking io ops to async method on init * [autofix.ci] apply automated fixes * use correct lib * imports --------- Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: Christophe Bornet <cbornet@hotmail.com>
This commit is contained in:
parent
03d64d21b4
commit
39b0d6dd1e
2 changed files with 12 additions and 7 deletions
|
|
@ -9,6 +9,7 @@ from datetime import datetime, timezone
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import TYPE_CHECKING
|
from typing import TYPE_CHECKING
|
||||||
|
|
||||||
|
import anyio
|
||||||
import sqlalchemy as sa
|
import sqlalchemy as sa
|
||||||
from alembic import command, util
|
from alembic import command, util
|
||||||
from alembic.config import Config
|
from alembic.config import Config
|
||||||
|
|
@ -43,31 +44,33 @@ class DatabaseService(Service):
|
||||||
raise ValueError(msg)
|
raise ValueError(msg)
|
||||||
self.database_url: str = settings_service.settings.database_url
|
self.database_url: str = settings_service.settings.database_url
|
||||||
self._sanitize_database_url()
|
self._sanitize_database_url()
|
||||||
|
|
||||||
# This file is in langflow.services.database.manager.py
|
# This file is in langflow.services.database.manager.py
|
||||||
# the ini is in langflow
|
# the ini is in langflow
|
||||||
langflow_dir = Path(__file__).parent.parent.parent
|
langflow_dir = Path(__file__).parent.parent.parent
|
||||||
self.script_location = langflow_dir / "alembic"
|
self.script_location = langflow_dir / "alembic"
|
||||||
self.alembic_cfg_path = langflow_dir / "alembic.ini"
|
self.alembic_cfg_path = langflow_dir / "alembic.ini"
|
||||||
|
|
||||||
# register the event listener for sqlite as part of this class.
|
# register the event listener for sqlite as part of this class.
|
||||||
# Using decorator will make the method not able to use self
|
# Using decorator will make the method not able to use self
|
||||||
event.listen(Engine, "connect", self.on_connection)
|
event.listen(Engine, "connect", self.on_connection)
|
||||||
self.engine = self._create_engine()
|
self.engine = self._create_engine()
|
||||||
self.async_engine = self._create_async_engine()
|
self.async_engine = self._create_async_engine()
|
||||||
alembic_log_file = self.settings_service.settings.alembic_log_file
|
|
||||||
|
|
||||||
|
alembic_log_file = self.settings_service.settings.alembic_log_file
|
||||||
# Check if the provided path is absolute, cross-platform.
|
# Check if the provided path is absolute, cross-platform.
|
||||||
if Path(alembic_log_file).is_absolute():
|
if Path(alembic_log_file).is_absolute():
|
||||||
# Use the absolute path directly.
|
|
||||||
self.alembic_log_path = Path(alembic_log_file)
|
self.alembic_log_path = Path(alembic_log_file)
|
||||||
else:
|
else:
|
||||||
# Construct the path using the langflow directory.
|
|
||||||
self.alembic_log_path = Path(langflow_dir) / alembic_log_file
|
self.alembic_log_path = Path(langflow_dir) / alembic_log_file
|
||||||
|
|
||||||
# Ensure the directory and file for the alembic log file exists
|
|
||||||
self.alembic_log_path.parent.mkdir(parents=True, exist_ok=True)
|
|
||||||
self.alembic_log_path.touch(exist_ok=True)
|
|
||||||
self._logged_pragma = False
|
self._logged_pragma = False
|
||||||
|
|
||||||
|
async def initialize_alembic_log_file(self):
|
||||||
|
# Ensure the directory and file for the alembic log file exists
|
||||||
|
anyio.Path(self.alembic_log_path.parent).mkdir(parents=True, exist_ok=True)
|
||||||
|
anyio.Path(self.alembic_log_path).touch(exist_ok=True)
|
||||||
|
|
||||||
def reload_engine(self) -> None:
|
def reload_engine(self) -> None:
|
||||||
self._sanitize_database_url()
|
self._sanitize_database_url()
|
||||||
self.engine = self._create_engine()
|
self.engine = self._create_engine()
|
||||||
|
|
|
||||||
|
|
@ -238,7 +238,9 @@ async def initialize_services(*, fix_migration: bool = False) -> None:
|
||||||
get_service(ServiceType.CACHE_SERVICE, default=CacheServiceFactory())
|
get_service(ServiceType.CACHE_SERVICE, default=CacheServiceFactory())
|
||||||
# Setup the superuser
|
# Setup the superuser
|
||||||
await initialize_database(fix_migration=fix_migration)
|
await initialize_database(fix_migration=fix_migration)
|
||||||
async with get_db_service().with_async_session() as session:
|
db_service = get_db_service()
|
||||||
|
await db_service.initialize_alembic_log_file()
|
||||||
|
async with db_service.with_async_session() as session:
|
||||||
settings_service = get_service(ServiceType.SETTINGS_SERVICE)
|
settings_service = get_service(ServiceType.SETTINGS_SERVICE)
|
||||||
await setup_superuser(settings_service, session)
|
await setup_superuser(settings_service, session)
|
||||||
try:
|
try:
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue