Repository: topoteretes/cognee
File: cognee/infrastructure/databases/relational/sqlalchemy/SqlAlchemyAdapter.py (Python)

Code before:
                or 'postgresql://user:pass@host:port/db').
            connect_args (dict, optional): Database driver arguments. These take precedence over
                DATABASE_CONNECT_ARGS environment variable.

        Environment Variables:
        ----------------------
            DATABASE_CONNECT_ARGS: Optional JSON string containing connection arguments.
                Allows configuration of driver-specific parameters such as SSL settings,
                timeouts, or connection pool options without code changes.

                Examples:
                    PostgreSQL with SSL:
                        DATABASE_CONNECT_ARGS='{"sslmode": "require", "connect_timeout": 10}'

                    SQLite with custom timeout:
                        DATABASE_CONNECT_ARGS='{"timeout": 60}'

                Note: This follows cognee's environment-based configuration pattern and is
                the recommended approach for production deployments.
        """
        self.db_path: str = None
        self.db_uri: str = connection_string

        # Parse optional connection arguments from environment variable
        env_connect_args_dict = {}
        env_connect_args = os.getenv("DATABASE_CONNECT_ARGS")
        if env_connect_args:
            try:
                parsed_args = json.loads(env_connect_args)
                if isinstance(parsed_args, dict):
                    env_connect_args_dict = parsed_args
                else:
                    logger.warning("DATABASE_CONNECT_ARGS is not a valid JSON dictionary, ignoring")
            except json.JSONDecodeError:
                logger.warning("Failed to parse DATABASE_CONNECT_ARGS as JSON, ignoring")

        # Merge environment args with explicit args (explicit args take precedence)
        final_connect_args = {**env_connect_args_dict, **(connect_args or {})}

        if "sqlite" in connection_string:
            [prefix, db_path] = connection_string.split("///")
            self.db_path = db_path

            if "s3://" in self.db_path:
                db_dir_path = path.dirname(self.db_path)
                file_storage = get_file_storage(db_dir_path)

                run_sync(file_storage.ensure_directory_exists())

                with tempfile.NamedTemporaryFile(mode="w", delete=False) as temp_file:
                    self.temp_db_file = temp_file.name

Review comment (suggestion):
We have a config for relational databases, called `RelationalConfig`. We should add the new parameters from the env variables there, that way they are loaded in the config, and we don't need to parse them here.

Code after:
            connect_args (dict, optional): Database driver connection arguments.
                Configuration is loaded from RelationalConfig.database_connect_args, which reads
                from the DATABASE_CONNECT_ARGS environment variable.

                Examples:
                    PostgreSQL with SSL:
                        DATABASE_CONNECT_ARGS='{"sslmode": "require", "connect_timeout": 10}'

                    SQLite with custom timeout:
                        DATABASE_CONNECT_ARGS='{"timeout": 60}'
        """
        self.db_path: str = None
        self.db_uri: str = connection_string

        # Use provided connect_args (already parsed from config)
        final_connect_args = connect_args or {}

        if "sqlite" in connection_string:
            [prefix, db_path] = connection_string.split("///")
            self.db_path = db_path

            if "s3://" in self.db_path:
                db_dir_path = path.dirname(self.db_path)
                file_storage = get_file_storage(db_dir_path)

                run_sync(file_storage.ensure_directory_exists())

                with tempfile.NamedTemporaryFile(mode="w", delete=False) as temp_file:
                    self.temp_db_file = temp_file.name
                    connection_string = prefix + "///" + self.temp_db_file

                run_sync(self.pull_from_s3())

        if "sqlite" in connection_string:
            self.engine = create_async_engine(
                connection_string,
                poolclass=NullPool,
                connect_args={**{"timeout": 30}, **final_connect_args},
            )
        else:
            self.engine = create_async_engine(
                connection_string,
                pool_size=20,
                max_overflow=20,
                pool_recycle=280,
                pool_pre_ping=True,
                pool_timeout=280,
                connect_args=final_connect_args,
            )

        self.sessionmaker = async_sessionmaker(bind=self.engine, expire_on_commit=False)