Skip to content

StarRocks

The starrocks/sqlalchemy adapter provides integration with StarRocks, a high-performance analytical database compatible with the MySQL protocol, using SQLAlchemy.

Session Managers

StarRocks-specific session manager handling connection management for the StarRocks analytical engine.

archipy.adapters.starrocks.sqlalchemy.session_managers.StarRocksSQlAlchemySessionManager

Bases: BaseSQLAlchemySessionManager[StarRocksSQLAlchemyConfig]

Synchronous SQLAlchemy session manager for StarRocks.

Inherits from BaseSQLAlchemySessionManager to provide StarRocks-specific session management, including connection URL creation and engine configuration.

Parameters:

Name Type Description Default
orm_config StarRocksSQLAlchemyConfig | None

StarRocks-specific configuration. If None, uses global config.

 None
Source code in archipy/adapters/starrocks/sqlalchemy/session_managers.py 
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
class StarRocksSQlAlchemySessionManager(BaseSQLAlchemySessionManager[StarRocksSQLAlchemyConfig], metaclass=Singleton):
    """Synchronous SQLAlchemy session manager for StarRocks.

    Inherits from BaseSQLAlchemySessionManager to provide StarRocks-specific session
    management, including connection URL creation and engine configuration.

    Args:
        orm_config: StarRocks-specific configuration. If None, uses global config.
    """

    def __init__(self, orm_config: StarRocksSQLAlchemyConfig | None = None) -> None:
        """Initialize the StarRocks session manager.

        Args:
            orm_config: StarRocks-specific configuration. If None, uses global config.
        """
        configs = BaseConfig.global_config().STARROCKS_SQLALCHEMY if orm_config is None else orm_config
        super().__init__(configs)

    @override
    def _expected_config_type(self) -> type[StarRocksSQLAlchemyConfig]:
        """Return the expected configuration type for StarRocks.

        Returns:
            The StarRocksSQLAlchemyConfig class.
        """
        return StarRocksSQLAlchemyConfig

    @override
    def _get_database_name(self) -> str:
        """Return the name of the database being used.

        Returns:
            str: The name of the database ('starrocks').
        """
        return "starrocks"

    @override
    def _get_connect_args(self) -> dict:
        """Return connection arguments for StarRocks to ensure proper transaction support.

        StarRocks (using MySQL protocol) requires autocommit to be explicitly disabled
        to ensure transactions work properly with rollback support.

        Returns:
            A dictionary with autocommit=False and connect_timeout from config.
        """
        connect_args = {}

        # Add connect_timeout if configured
        if hasattr(self, "_configs"):
            if hasattr(self._configs, "CONNECT_TIMEOUT") and self._configs.CONNECT_TIMEOUT is not None:
                connect_args["connect_timeout"] = self._configs.CONNECT_TIMEOUT

        # Add StarRocks-specific setting for transaction support
        connect_args["autocommit"] = False

        return connect_args

    @override
    def _create_url(self, configs: StarRocksSQLAlchemyConfig) -> URL:
        """Create a StarRocks connection URL.

        Args:
            configs: StarRocks configuration.

        Returns:
            A SQLAlchemy URL object for StarRocks.

        Raises:
            DatabaseConnectionError: If there's an error creating the URL.
        """
        try:
            return URL.create(
                drivername=configs.DRIVER_NAME,
                username=configs.USERNAME,
                password=configs.PASSWORD,
                host=configs.HOST,
                port=configs.PORT,
                database=configs.DATABASE,
            )
        except SQLAlchemyError as e:
            raise DatabaseConnectionError(
                database=self._get_database_name(),
            ) from e

archipy.adapters.starrocks.sqlalchemy.session_managers.StarRocksSQlAlchemySessionManager.engine instance-attribute 

engine = _create_engine(orm_config)

archipy.adapters.starrocks.sqlalchemy.session_managers.StarRocksSQlAlchemySessionManager.get_session 

get_session() -> Session

Retrieve a thread-safe SQLAlchemy session.

Returns:

Name Type Description
Session Session

A SQLAlchemy session instance for database operations.

Raises:

Type Description
DatabaseConnectionError

If there's an error creating the session.
DatabaseConfigurationError

If there's an error in the database configuration.
Source code in archipy/adapters/base/sqlalchemy/session_managers.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
@override
def get_session(self) -> Session:
    """Retrieve a thread-safe SQLAlchemy session.

    Returns:
        Session: A SQLAlchemy session instance for database operations.

    Raises:
        DatabaseConnectionError: If there's an error creating the session.
        DatabaseConfigurationError: If there's an error in the database configuration.
    """
    try:
        return self._session_generator()
    except SQLAlchemyError as e:
        if "configuration" in str(e).lower():
            raise DatabaseConfigurationError(
                database=self._get_database_name(),
            ) from e
        raise DatabaseConnectionError(
            database=self._get_database_name(),
        ) from e

archipy.adapters.starrocks.sqlalchemy.session_managers.StarRocksSQlAlchemySessionManager.remove_session 

remove_session() -> None

Remove the current session from the registry.

Cleans up the session to prevent resource leaks, typically called at the end of a request.

Raises:

Type Description
DatabaseConnectionError

If there's an error removing the session.
DatabaseConfigurationError

If there's an error in the database configuration.
Source code in archipy/adapters/base/sqlalchemy/session_managers.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
@override
def remove_session(self) -> None:
    """Remove the current session from the registry.

    Cleans up the session to prevent resource leaks, typically called at the end
    of a request.

    Raises:
        DatabaseConnectionError: If there's an error removing the session.
        DatabaseConfigurationError: If there's an error in the database configuration.
    """
    try:
        self._session_generator.remove()
    except SQLAlchemyError as e:
        if "configuration" in str(e).lower():
            raise DatabaseConfigurationError(
                database=self._get_database_name(),
            ) from e
        raise DatabaseConnectionError(
            database=self._get_database_name(),
        ) from e

archipy.adapters.starrocks.sqlalchemy.session_managers.AsyncStarRocksSQlAlchemySessionManager

Bases: AsyncBaseSQLAlchemySessionManager[StarRocksSQLAlchemyConfig]

Asynchronous SQLAlchemy session manager for StarRocks.

Inherits from AsyncBaseSQLAlchemySessionManager to provide async StarRocks-specific session management, including connection URL creation and async engine configuration.

Parameters:

Name Type Description Default
orm_config StarRocksSQLAlchemyConfig | None

StarRocks-specific configuration. If None, uses global config.

 None
Source code in archipy/adapters/starrocks/sqlalchemy/session_managers.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
class AsyncStarRocksSQlAlchemySessionManager(
    AsyncBaseSQLAlchemySessionManager[StarRocksSQLAlchemyConfig],
    metaclass=Singleton,
):
    """Asynchronous SQLAlchemy session manager for StarRocks.

    Inherits from AsyncBaseSQLAlchemySessionManager to provide async StarRocks-specific
    session management, including connection URL creation and async engine configuration.

    Args:
        orm_config: StarRocks-specific configuration. If None, uses global config.
    """

    def __init__(self, orm_config: StarRocksSQLAlchemyConfig | None = None) -> None:
        """Initialize the async StarRocks session manager.

        Args:
            orm_config: StarRocks-specific configuration. If None, uses global config.
        """
        configs = BaseConfig.global_config().STARROCKS_SQLALCHEMY if orm_config is None else orm_config
        super().__init__(configs)

    @override
    def _expected_config_type(self) -> type[StarRocksSQLAlchemyConfig]:
        """Return the expected configuration type for StarRocks.

        Returns:
            The StarRocksSQLAlchemyConfig class.
        """
        return StarRocksSQLAlchemyConfig

    @override
    def _get_database_name(self) -> str:
        """Return the name of the database being used.

        Returns:
            str: The name of the database ('starrocks').
        """
        return "starrocks"

    @override
    def _create_url(self, configs: StarRocksSQLAlchemyConfig) -> URL:
        """Create an async StarRocks connection URL.

        For async operations, StarRocks requires the starrocks+asyncmy driver
        which uses the asyncmy library for async MySQL protocol support while
        maintaining StarRocks dialect features (type mapping, compiler patches).

        Args:
            configs: StarRocks configuration.

        Returns:
            A SQLAlchemy URL object for StarRocks with async driver.

        Raises:
            DatabaseConnectionError: If there's an error creating the URL.
        """
        try:
            return URL.create(
                drivername="starrocks+asyncmy",
                username=configs.USERNAME,
                password=configs.PASSWORD,
                host=configs.HOST,
                port=configs.PORT,
                database=configs.DATABASE,
            )
        except SQLAlchemyError as e:
            raise DatabaseConnectionError(
                database=self._get_database_name(),
            ) from e

    @override
    def _get_connect_args(self) -> dict:
        """Return connection arguments for async StarRocks to ensure proper transaction support.

        StarRocks (using MySQL protocol via asyncmy) requires autocommit to be explicitly disabled
        to ensure transactions work properly with rollback support.

        Note: asyncmy driver only supports connect_timeout, not read_timeout/write_timeout.
        These socket-level timeouts are handled differently in async drivers.

        Returns:
            A dictionary with autocommit=False and connect_timeout (no read/write timeouts for asyncmy).
        """
        connect_args = {}

        # Add connect_timeout if configured
        if hasattr(self, "_configs"):
            if hasattr(self._configs, "CONNECT_TIMEOUT") and self._configs.CONNECT_TIMEOUT is not None:
                connect_args["connect_timeout"] = self._configs.CONNECT_TIMEOUT

        # Add StarRocks async-specific setting for transaction support
        connect_args["autocommit"] = False

        return connect_args

archipy.adapters.starrocks.sqlalchemy.session_managers.AsyncStarRocksSQlAlchemySessionManager.engine instance-attribute 

engine = _create_async_engine(orm_config)

archipy.adapters.starrocks.sqlalchemy.session_managers.AsyncStarRocksSQlAlchemySessionManager.get_session 

get_session() -> AsyncSession

Retrieve a task-safe async SQLAlchemy session.

Returns:

Name Type Description
AsyncSession AsyncSession

An async SQLAlchemy session instance for database operations.

Raises:

Type Description
DatabaseConnectionError

If there's an error creating the session.
DatabaseConfigurationError

If there's an error in the database configuration.
Source code in archipy/adapters/base/sqlalchemy/session_managers.py 
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
@override
def get_session(self) -> AsyncSession:
    """Retrieve a task-safe async SQLAlchemy session.

    Returns:
        AsyncSession: An async SQLAlchemy session instance for database operations.

    Raises:
        DatabaseConnectionError: If there's an error creating the session.
        DatabaseConfigurationError: If there's an error in the database configuration.
    """
    try:
        return self._session_generator()
    except SQLAlchemyError as e:
        if "configuration" in str(e).lower():
            raise DatabaseConfigurationError(
                database=self._get_database_name(),
            ) from e
        raise DatabaseConnectionError(
            database=self._get_database_name(),
        ) from e

archipy.adapters.starrocks.sqlalchemy.session_managers.AsyncStarRocksSQlAlchemySessionManager.remove_session async 

remove_session() -> None

Remove the current session from the registry.

Cleans up the session to prevent resource leaks, typically called at the end of a request.

Raises:

Type Description
DatabaseConnectionError

If there's an error removing the session.
DatabaseConfigurationError

If there's an error in the database configuration.
Source code in archipy/adapters/base/sqlalchemy/session_managers.py 
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
@override
async def remove_session(self) -> None:
    """Remove the current session from the registry.

    Cleans up the session to prevent resource leaks, typically called at the end
    of a request.

    Raises:
        DatabaseConnectionError: If there's an error removing the session.
        DatabaseConfigurationError: If there's an error in the database configuration.
    """
    try:
        await self._session_generator.remove()
    except SQLAlchemyError as e:
        if "configuration" in str(e).lower():
            raise DatabaseConfigurationError(
                database=self._get_database_name(),
            ) from e
        raise DatabaseConnectionError(
            database=self._get_database_name(),
        ) from e

options: show_root_toc_entry: false heading_level: 3

Session Manager Registry

Registry for StarRocks session manager instances.

archipy.adapters.starrocks.sqlalchemy.session_manager_registry.StarRocksSessionManagerRegistry

Bases: SessionManagerRegistry

Registry for StarRocks SQLAlchemy session managers.

This registry provides a centralized access point for both synchronous and asynchronous StarRocks session managers, implementing the Service Locator pattern. It lazily initializes the appropriate session manager when first requested.

The registry maintains singleton instances of: - A synchronous session manager (StarRocksSQlAlchemySessionManager) - An asynchronous session manager (AsyncStarRocksSQlAlchemySessionManager)

Source code in archipy/adapters/starrocks/sqlalchemy/session_manager_registry.py 
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
class StarRocksSessionManagerRegistry(SessionManagerRegistry, metaclass=Singleton):
    """Registry for StarRocks SQLAlchemy session managers.

    This registry provides a centralized access point for both synchronous and
    asynchronous StarRocks session managers, implementing the Service Locator pattern.
    It lazily initializes the appropriate session manager when first requested.

    The registry maintains singleton instances of:
    - A synchronous session manager (StarRocksSQlAlchemySessionManager)
    - An asynchronous session manager (AsyncStarRocksSQlAlchemySessionManager)
    """

    @classmethod
    def get_sync_manager(cls) -> SessionManagerPort:
        """Get the synchronous StarRocks session manager instance.

        Lazily initializes a default StarRocksSQlAlchemySessionManager if none has been set.

        Returns:
            SessionManagerPort: The registered synchronous session manager

        Raises:
            DatabaseConnectionError: If there's an error initializing the session manager
        """
        if cls._sync_instance is None:
            try:
                from archipy.adapters.starrocks.sqlalchemy.session_managers import StarRocksSQlAlchemySessionManager

                cls._sync_instance = StarRocksSQlAlchemySessionManager()
            except Exception as e:
                raise DatabaseConnectionError(
                    database="starrocks",
                ) from e
        return cls._sync_instance

    @classmethod
    def set_sync_manager(cls, manager: SessionManagerPort) -> None:
        """Set a custom synchronous session manager.

        Args:
            manager: An instance implementing SessionManagerPort

        Raises:
            InvalidArgumentError: If the manager is None or doesn't implement SessionManagerPort
        """
        if manager is None:
            raise InvalidArgumentError("StarRocks session manager cannot be None")
        from archipy.adapters.base.sqlalchemy.session_manager_ports import SessionManagerPort

        if not isinstance(manager, SessionManagerPort):
            raise InvalidArgumentError(f"Manager must implement SessionManagerPort, got {type(manager).__name__}")
        cls._sync_instance = manager

    @classmethod
    def get_async_manager(cls) -> AsyncSessionManagerPort:
        """Get the asynchronous StarRocks session manager instance.

        Lazily initializes a default AsyncStarRocksSQlAlchemySessionManager if none has been set.

        Returns:
            AsyncSessionManagerPort: The registered asynchronous session manager

        Raises:
            DatabaseConnectionError: If there's an error initializing the session manager
        """
        if cls._async_instance is None:
            try:
                from archipy.adapters.starrocks.sqlalchemy.session_managers import (
                    AsyncStarRocksSQlAlchemySessionManager,
                )

                cls._async_instance = AsyncStarRocksSQlAlchemySessionManager()
            except Exception as e:
                raise DatabaseConnectionError(
                    database="starrocks",
                ) from e
        return cls._async_instance

    @classmethod
    def set_async_manager(cls, manager: AsyncSessionManagerPort) -> None:
        """Set a custom asynchronous session manager.

        Args:
            manager: An instance implementing AsyncSessionManagerPort

        Raises:
            InvalidArgumentError: If the manager is None or doesn't implement AsyncSessionManagerPort
        """
        if manager is None:
            raise InvalidArgumentError("StarRocks async session manager cannot be None")
        from archipy.adapters.base.sqlalchemy.session_manager_ports import AsyncSessionManagerPort

        if not isinstance(manager, AsyncSessionManagerPort):
            raise InvalidArgumentError(f"Manager must implement AsyncSessionManagerPort, got {type(manager).__name__}")
        cls._async_instance = manager

    @classmethod
    def reset(cls) -> None:
        """Reset the registry to its initial state.

        This method clears both registered managers, useful for testing.
        """
        cls._sync_instance = None
        cls._async_instance = None

archipy.adapters.starrocks.sqlalchemy.session_manager_registry.StarRocksSessionManagerRegistry.get_sync_manager classmethod 

get_sync_manager() -> SessionManagerPort

Get the synchronous StarRocks session manager instance.

Lazily initializes a default StarRocksSQlAlchemySessionManager if none has been set.

Returns:

Name Type Description
SessionManagerPort SessionManagerPort

The registered synchronous session manager

Raises:

Type Description
DatabaseConnectionError

If there's an error initializing the session manager
Source code in archipy/adapters/starrocks/sqlalchemy/session_manager_registry.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
@classmethod
def get_sync_manager(cls) -> SessionManagerPort:
    """Get the synchronous StarRocks session manager instance.

    Lazily initializes a default StarRocksSQlAlchemySessionManager if none has been set.

    Returns:
        SessionManagerPort: The registered synchronous session manager

    Raises:
        DatabaseConnectionError: If there's an error initializing the session manager
    """
    if cls._sync_instance is None:
        try:
            from archipy.adapters.starrocks.sqlalchemy.session_managers import StarRocksSQlAlchemySessionManager

            cls._sync_instance = StarRocksSQlAlchemySessionManager()
        except Exception as e:
            raise DatabaseConnectionError(
                database="starrocks",
            ) from e
    return cls._sync_instance

archipy.adapters.starrocks.sqlalchemy.session_manager_registry.StarRocksSessionManagerRegistry.set_sync_manager classmethod 

set_sync_manager(manager: SessionManagerPort) -> None

Set a custom synchronous session manager.

Parameters:

Name Type Description Default
manager SessionManagerPort

An instance implementing SessionManagerPort

 required

Raises:

Type Description
InvalidArgumentError

If the manager is None or doesn't implement SessionManagerPort
Source code in archipy/adapters/starrocks/sqlalchemy/session_manager_registry.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
@classmethod
def set_sync_manager(cls, manager: SessionManagerPort) -> None:
    """Set a custom synchronous session manager.

    Args:
        manager: An instance implementing SessionManagerPort

    Raises:
        InvalidArgumentError: If the manager is None or doesn't implement SessionManagerPort
    """
    if manager is None:
        raise InvalidArgumentError("StarRocks session manager cannot be None")
    from archipy.adapters.base.sqlalchemy.session_manager_ports import SessionManagerPort

    if not isinstance(manager, SessionManagerPort):
        raise InvalidArgumentError(f"Manager must implement SessionManagerPort, got {type(manager).__name__}")
    cls._sync_instance = manager

archipy.adapters.starrocks.sqlalchemy.session_manager_registry.StarRocksSessionManagerRegistry.get_async_manager classmethod 

get_async_manager() -> AsyncSessionManagerPort

Get the asynchronous StarRocks session manager instance.

Lazily initializes a default AsyncStarRocksSQlAlchemySessionManager if none has been set.

Returns:

Name Type Description
AsyncSessionManagerPort AsyncSessionManagerPort

The registered asynchronous session manager

Raises:

Type Description
DatabaseConnectionError

If there's an error initializing the session manager
Source code in archipy/adapters/starrocks/sqlalchemy/session_manager_registry.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@classmethod
def get_async_manager(cls) -> AsyncSessionManagerPort:
    """Get the asynchronous StarRocks session manager instance.

    Lazily initializes a default AsyncStarRocksSQlAlchemySessionManager if none has been set.

    Returns:
        AsyncSessionManagerPort: The registered asynchronous session manager

    Raises:
        DatabaseConnectionError: If there's an error initializing the session manager
    """
    if cls._async_instance is None:
        try:
            from archipy.adapters.starrocks.sqlalchemy.session_managers import (
                AsyncStarRocksSQlAlchemySessionManager,
            )

            cls._async_instance = AsyncStarRocksSQlAlchemySessionManager()
        except Exception as e:
            raise DatabaseConnectionError(
                database="starrocks",
            ) from e
    return cls._async_instance

archipy.adapters.starrocks.sqlalchemy.session_manager_registry.StarRocksSessionManagerRegistry.set_async_manager classmethod 

set_async_manager(manager: AsyncSessionManagerPort) -> None

Set a custom asynchronous session manager.

Parameters:

Name Type Description Default
manager AsyncSessionManagerPort

An instance implementing AsyncSessionManagerPort

 required

Raises:

Type Description
InvalidArgumentError

If the manager is None or doesn't implement AsyncSessionManagerPort
Source code in archipy/adapters/starrocks/sqlalchemy/session_manager_registry.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
@classmethod
def set_async_manager(cls, manager: AsyncSessionManagerPort) -> None:
    """Set a custom asynchronous session manager.

    Args:
        manager: An instance implementing AsyncSessionManagerPort

    Raises:
        InvalidArgumentError: If the manager is None or doesn't implement AsyncSessionManagerPort
    """
    if manager is None:
        raise InvalidArgumentError("StarRocks async session manager cannot be None")
    from archipy.adapters.base.sqlalchemy.session_manager_ports import AsyncSessionManagerPort

    if not isinstance(manager, AsyncSessionManagerPort):
        raise InvalidArgumentError(f"Manager must implement AsyncSessionManagerPort, got {type(manager).__name__}")
    cls._async_instance = manager

archipy.adapters.starrocks.sqlalchemy.session_manager_registry.StarRocksSessionManagerRegistry.reset classmethod 

reset() -> None

Reset the registry to its initial state.

This method clears both registered managers, useful for testing.

Source code in archipy/adapters/starrocks/sqlalchemy/session_manager_registry.py 
107
108
109
110
111
112
113
114
@classmethod
def reset(cls) -> None:
    """Reset the registry to its initial state.

    This method clears both registered managers, useful for testing.
    """
    cls._sync_instance = None
    cls._async_instance = None

options: show_root_toc_entry: false heading_level: 3

Adapters

Concrete StarRocks adapter built on the base SQLAlchemy adapter with StarRocks-specific dialect configuration.

archipy.adapters.starrocks.sqlalchemy.adapters.StarrocksSQLAlchemyAdapter

Bases: BaseSQLAlchemyAdapter[StarRocksSQLAlchemyConfig]

Synchronous SQLAlchemy adapter for Starrocks.

Inherits from BaseSQLAlchemyAdapter to provide Starrocks-specific session management and database operations.

Parameters:

Name Type Description Default
orm_config StarRocksSQLAlchemyConfig | None

Starrocks-specific configuration. If None, uses global config.

 None
Source code in archipy/adapters/starrocks/sqlalchemy/adapters.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
class StarrocksSQLAlchemyAdapter(BaseSQLAlchemyAdapter[StarRocksSQLAlchemyConfig]):
    """Synchronous SQLAlchemy adapter for Starrocks.

    Inherits from BaseSQLAlchemyAdapter to provide Starrocks-specific session management
    and database operations.

    Args:
        orm_config: Starrocks-specific configuration. If None, uses global config.
    """

    def __init__(self, orm_config: StarRocksSQLAlchemyConfig | None = None) -> None:
        """Initialize the Starrocks adapter with a session manager.

        Args:
            orm_config: Starrocks-specific configuration. If None, uses global config.
        """
        configs = BaseConfig.global_config().STARROCKS_SQLALCHEMY if orm_config is None else orm_config
        super().__init__(configs)

    @override
    def _create_session_manager(self, configs: StarRocksSQLAlchemyConfig) -> StarRocksSQlAlchemySessionManager:
        """Create a Starrocks-specific session manager.

        Args:
            configs: Starrocks configuration.

        Returns:
            A Starrocks session manager instance.
        """
        return StarRocksSQlAlchemySessionManager(configs)

archipy.adapters.starrocks.sqlalchemy.adapters.StarrocksSQLAlchemyAdapter.session_manager instance-attribute 

session_manager: BaseSQLAlchemySessionManager[ConfigT] = (
    _create_session_manager(configs)
)

archipy.adapters.starrocks.sqlalchemy.adapters.StarrocksSQLAlchemyAdapter.get_session 

get_session() -> Session

Get a database session.

Returns:

Name Type Description
Session Session

A SQLAlchemy session.

Raises:

Type Description
DatabaseConnectionError

If there's an error getting the session.
DatabaseConfigurationError

If there's an error in the database configuration.
Source code in archipy/adapters/base/sqlalchemy/adapters.py 
308
309
310
311
312
313
314
315
316
317
318
319
@override
def get_session(self) -> Session:
    """Get a database session.

    Returns:
        Session: A SQLAlchemy session.

    Raises:
        DatabaseConnectionError: If there's an error getting the session.
        DatabaseConfigurationError: If there's an error in the database configuration.
    """
    return self.session_manager.get_session()

archipy.adapters.starrocks.sqlalchemy.adapters.StarrocksSQLAlchemyAdapter.execute_search_query 

execute_search_query(
    entity: type[T],
    query: Select,
    pagination: PaginationDTO | None = None,
    sort_info: SortDTO | None = None,
    has_multiple_entities: bool = False,
) -> tuple[list[T], int]

Execute a search query with pagination and sorting.

Parameters:

Name Type Description Default
entity type[T]

The entity class to query.

 required
query Select

The SQLAlchemy SELECT query.

 required
pagination PaginationDTO | None

Optional pagination settings.

 None
sort_info SortDTO | None

Optional sorting information.

 None
has_multiple_entities bool

Optional bool.

 False

Returns:

Type Description
tuple[list[T], int]

Tuple of the list of entities and the total count.

Raises:

Type Description
DatabaseQueryError

If the database query fails.
DatabaseTimeoutError

If the query times out.
DatabaseConnectionError

If there's a connection error.
DatabaseTransactionError

If there's a transaction error.
Source code in archipy/adapters/base/sqlalchemy/adapters.py 
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
@override
def execute_search_query(
    self,
    entity: type[T],
    query: Select,
    pagination: PaginationDTO | None = None,
    sort_info: SortDTO | None = None,
    has_multiple_entities: bool = False,
) -> tuple[list[T], int]:
    """Execute a search query with pagination and sorting.

    Args:
        entity: The entity class to query.
        query: The SQLAlchemy SELECT query.
        pagination: Optional pagination settings.
        sort_info: Optional sorting information.
        has_multiple_entities: Optional bool.

    Returns:
        Tuple of the list of entities and the total count.

    Raises:
        DatabaseQueryError: If the database query fails.
        DatabaseTimeoutError: If the query times out.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    try:
        sort_info = sort_info or SortDTO.default()
        session = self.get_session()
        sorted_query = self._apply_sorting(entity, query, sort_info)
        paginated_query = self._apply_pagination(sorted_query, pagination)
        result_set = session.execute(paginated_query)
        if has_multiple_entities:
            # For multiple entities, fetchall returns list of Row objects
            raw_results = list(result_set.fetchall())
            # Convert to list[T] - each row contains entities of type T
            # Use tuple unpacking to access the first element
            results: list[T] = []
            for row in raw_results:
                if row:
                    # Row supports indexing and tuple unpacking
                    row_tuple = tuple(row)
                    if row_tuple:
                        first_entity = row_tuple[0]
                        # first_entity is T (entity type), verify it's an instance
                        if isinstance(first_entity, entity):
                            results.append(first_entity)
        else:
            # For single entity, scalars() returns list[T] directly
            results = list(result_set.scalars().all())
        count_query = select(func.count()).select_from(query.subquery())
        total_count = session.execute(count_query).scalar_one()
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())
        raise  # This will never be reached, but satisfies MyPy
    else:
        # Type: results is list[T] where T extends BaseEntity, total_count is int
        return results, total_count

archipy.adapters.starrocks.sqlalchemy.adapters.StarrocksSQLAlchemyAdapter.create 

create(entity: T) -> T | None

Create a new entity in the database.

Parameters:

Name Type Description Default
entity T

The entity to create.

 required

Returns:

Type Description
T | None

The created entity with updated attributes, preserving the original type.

Raises:

Type Description
InvalidEntityTypeError

If the entity type is not a valid SQLAlchemy model.
DatabaseQueryError

If the database operation fails.
DatabaseIntegrityError

If there's an integrity constraint violation.
DatabaseConstraintError

If there's a constraint violation.
DatabaseConnectionError

If there's a connection error.
DatabaseTransactionError

If there's a transaction error.
Source code in archipy/adapters/base/sqlalchemy/adapters.py 
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
@override
def create(self, entity: T) -> T | None:
    """Create a new entity in the database.

    Args:
        entity: The entity to create.

    Returns:
        The created entity with updated attributes, preserving the original type.

    Raises:
        InvalidEntityTypeError: If the entity type is not a valid SQLAlchemy model.
        DatabaseQueryError: If the database operation fails.
        DatabaseIntegrityError: If there's an integrity constraint violation.
        DatabaseConstraintError: If there's a constraint violation.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    if not isinstance(entity, BaseEntity):
        raise InvalidEntityTypeError(
            message=f"Expected BaseEntity subclass, got {type(entity).__name__}",
            expected_type="BaseEntity",
            actual_type=type(entity).__name__,
        )

    try:
        session = self.get_session()
        session.add(entity)
        session.flush()
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())
        raise  # This will never be reached, but satisfies MyPy
    else:
        return entity

archipy.adapters.starrocks.sqlalchemy.adapters.StarrocksSQLAlchemyAdapter.bulk_create 

bulk_create(entities: list[T]) -> list[T] | None

Creates multiple entities in a single database operation.

Parameters:

Name Type Description Default
entities list[T]

List of entities to create.

 required

Returns:

Type Description
list[T] | None

List of created entities with updated attributes, preserving original types.

Raises:

Type Description
InvalidEntityTypeError

If any entity is not a valid SQLAlchemy model.
DatabaseQueryError

If the database operation fails.
DatabaseIntegrityError

If there's an integrity constraint violation.
DatabaseConstraintError

If there's a constraint violation.
DatabaseConnectionError

If there's a connection error.
DatabaseTransactionError

If there's a transaction error.
Source code in archipy/adapters/base/sqlalchemy/adapters.py 
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
@override
def bulk_create(self, entities: list[T]) -> list[T] | None:
    """Creates multiple entities in a single database operation.

    Args:
        entities: List of entities to create.

    Returns:
        List of created entities with updated attributes, preserving original types.

    Raises:
        InvalidEntityTypeError: If any entity is not a valid SQLAlchemy model.
        DatabaseQueryError: If the database operation fails.
        DatabaseIntegrityError: If there's an integrity constraint violation.
        DatabaseConstraintError: If there's a constraint violation.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    if not all(isinstance(entity, BaseEntity) for entity in entities):
        raise InvalidEntityTypeError(
            message="All entities must be BaseEntity subclasses",
            expected_type="BaseEntity",
            actual_type="mixed",
        )

    try:
        session = self.get_session()
        session.add_all(entities)
        session.flush()
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())
        raise  # This will never be reached, but satisfies MyPy
    else:
        return entities

archipy.adapters.starrocks.sqlalchemy.adapters.StarrocksSQLAlchemyAdapter.get_by_uuid 

get_by_uuid(
    entity_type: type[T], entity_uuid: UUID
) -> BaseEntity | None

Retrieve an entity by its UUID.

Parameters:

Name Type Description Default
entity_type type[T]

The type of entity to retrieve.

 required
entity_uuid UUID

The UUID of the entity.

 required

Returns:

Type Description
BaseEntity | None

The entity if found, None otherwise.

Raises:

Type Description
InvalidEntityTypeError

If the entity type is not a valid SQLAlchemy model.
DatabaseQueryError

If the database operation fails.
DatabaseTimeoutError

If the query times out.
DatabaseConnectionError

If there's a connection error.
DatabaseTransactionError

If there's a transaction error.
Source code in archipy/adapters/base/sqlalchemy/adapters.py 
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
@override
def get_by_uuid(self, entity_type: type[T], entity_uuid: UUID) -> BaseEntity | None:
    """Retrieve an entity by its UUID.

    Args:
        entity_type: The type of entity to retrieve.
        entity_uuid: The UUID of the entity.

    Returns:
        The entity if found, None otherwise.

    Raises:
        InvalidEntityTypeError: If the entity type is not a valid SQLAlchemy model.
        DatabaseQueryError: If the database operation fails.
        DatabaseTimeoutError: If the query times out.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    if not issubclass(entity_type, BaseEntity):
        raise InvalidEntityTypeError(
            message=f"Expected BaseEntity subclass, got {entity_type.__name__}",
            expected_type="BaseEntity",
            actual_type=entity_type.__name__,
        )

    try:
        session = self.get_session()
        result = session.get(entity_type, entity_uuid)
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())
        raise  # This will never be reached, but satisfies MyPy
    else:
        # result is T | None where T extends BaseEntity, compatible with BaseEntity | None
        # The type checker needs explicit type annotation to understand the relationship
        typed_result: BaseEntity | None = result
        return typed_result

archipy.adapters.starrocks.sqlalchemy.adapters.StarrocksSQLAlchemyAdapter.delete 

delete(entity: T) -> None

Delete an entity from the database.

Parameters:

Name Type Description Default
entity T

The entity to delete.

 required

Raises:

Type Description
InvalidEntityTypeError

If the entity is not a valid SQLAlchemy model.
DatabaseQueryError

If the database operation fails.
DatabaseIntegrityError

If there's an integrity constraint violation.
DatabaseConstraintError

If there's a constraint violation.
DatabaseConnectionError

If there's a connection error.
DatabaseTransactionError

If there's a transaction error.
Source code in archipy/adapters/base/sqlalchemy/adapters.py 
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
@override
def delete(self, entity: T) -> None:
    """Delete an entity from the database.

    Args:
        entity: The entity to delete.

    Raises:
        InvalidEntityTypeError: If the entity is not a valid SQLAlchemy model.
        DatabaseQueryError: If the database operation fails.
        DatabaseIntegrityError: If there's an integrity constraint violation.
        DatabaseConstraintError: If there's a constraint violation.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    if not isinstance(entity, BaseEntity):
        raise InvalidEntityTypeError(
            message=f"Expected BaseEntity subclass, got {type(entity).__name__}",
            expected_type="BaseEntity",
            actual_type=type(entity).__name__,
        )

    try:
        session = self.get_session()
        session.delete(entity)
        session.flush()
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())

archipy.adapters.starrocks.sqlalchemy.adapters.StarrocksSQLAlchemyAdapter.bulk_delete 

bulk_delete(entities: list[T]) -> None

Delete multiple entities from the database.

Parameters:

Name Type Description Default
entities list[T]

List of entities to delete.

 required

Raises:

Type Description
InvalidEntityTypeError

If any entity is not a valid SQLAlchemy model.
DatabaseQueryError

If the database operation fails.
DatabaseIntegrityError

If there's an integrity constraint violation.
DatabaseConstraintError

If there's a constraint violation.
DatabaseConnectionError

If there's a connection error.
DatabaseTransactionError

If there's a transaction error.
Source code in archipy/adapters/base/sqlalchemy/adapters.py 
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
@override
def bulk_delete(self, entities: list[T]) -> None:
    """Delete multiple entities from the database.

    Args:
        entities: List of entities to delete.

    Raises:
        InvalidEntityTypeError: If any entity is not a valid SQLAlchemy model.
        DatabaseQueryError: If the database operation fails.
        DatabaseIntegrityError: If there's an integrity constraint violation.
        DatabaseConstraintError: If there's a constraint violation.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    if not all(isinstance(entity, BaseEntity) for entity in entities):
        raise InvalidEntityTypeError(
            message="All entities must be BaseEntity subclasses",
            expected_type="BaseEntity",
            actual_type="mixed",
        )

    try:
        session = self.get_session()
        for entity in entities:
            session.delete(entity)
        session.flush()
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())

archipy.adapters.starrocks.sqlalchemy.adapters.StarrocksSQLAlchemyAdapter.execute 

execute(
    statement: Executable,
    params: AnyExecuteParams | None = None,
) -> Result[Any]

Execute a SQLAlchemy statement.

Parameters:

Name Type Description Default
statement Executable

The SQLAlchemy statement to execute.

 required
params AnyExecuteParams | None

Optional parameters for the statement.

 None

Returns:

Type Description
Result[Any]

The result of the execution.

Raises:

Type Description
DatabaseQueryError

If the database operation fails.
DatabaseTimeoutError

If the query times out.
DatabaseConnectionError

If there's a connection error.
DatabaseTransactionError

If there's a transaction error.
Source code in archipy/adapters/base/sqlalchemy/adapters.py 
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
@override
def execute(self, statement: Executable, params: AnyExecuteParams | None = None) -> Result[Any]:
    """Execute a SQLAlchemy statement.

    Args:
        statement: The SQLAlchemy statement to execute.
        params: Optional parameters for the statement.

    Returns:
        The result of the execution.

    Raises:
        DatabaseQueryError: If the database operation fails.
        DatabaseTimeoutError: If the query times out.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    try:
        session = self.get_session()
        result = session.execute(statement, params or {})
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())
        raise  # This will never be reached, but satisfies MyPy
    else:
        return result

archipy.adapters.starrocks.sqlalchemy.adapters.StarrocksSQLAlchemyAdapter.scalars 

scalars(
    statement: Executable,
    params: AnyExecuteParams | None = None,
) -> ScalarResult[Any]

Execute a SQLAlchemy statement and return scalar results.

Parameters:

Name Type Description Default
statement Executable

The SQLAlchemy statement to execute.

 required
params AnyExecuteParams | None

Optional parameters for the statement.

 None

Returns:

Type Description
ScalarResult[Any]

The scalar results of the execution.

Raises:

Type Description
DatabaseQueryError

If the database operation fails.
DatabaseTimeoutError

If the query times out.
DatabaseConnectionError

If there's a connection error.
DatabaseTransactionError

If there's a transaction error.
Source code in archipy/adapters/base/sqlalchemy/adapters.py 
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
@override
def scalars(self, statement: Executable, params: AnyExecuteParams | None = None) -> ScalarResult[Any]:
    """Execute a SQLAlchemy statement and return scalar results.

    Args:
        statement: The SQLAlchemy statement to execute.
        params: Optional parameters for the statement.

    Returns:
        The scalar results of the execution.

    Raises:
        DatabaseQueryError: If the database operation fails.
        DatabaseTimeoutError: If the query times out.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    try:
        session = self.get_session()
        result = session.scalars(statement, params or {})
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())
        raise  # This will never be reached, but satisfies MyPy
    else:
        return result

archipy.adapters.starrocks.sqlalchemy.adapters.AsyncStarrocksSQLAlchemyAdapter

Bases: AsyncBaseSQLAlchemyAdapter[StarRocksSQLAlchemyConfig]

Asynchronous SQLAlchemy adapter for Starrocks.

Inherits from AsyncBaseSQLAlchemyAdapter to provide async Starrocks-specific session management and database operations.

Parameters:

Name Type Description Default
orm_config StarRocksSQLAlchemyConfig | None

Starrocks-specific configuration. If None, uses global config.

 None
Source code in archipy/adapters/starrocks/sqlalchemy/adapters.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
class AsyncStarrocksSQLAlchemyAdapter(AsyncBaseSQLAlchemyAdapter[StarRocksSQLAlchemyConfig]):
    """Asynchronous SQLAlchemy adapter for Starrocks.

    Inherits from AsyncBaseSQLAlchemyAdapter to provide async Starrocks-specific session
    management and database operations.

    Args:
        orm_config: Starrocks-specific configuration. If None, uses global config.
    """

    def __init__(self, orm_config: StarRocksSQLAlchemyConfig | None = None) -> None:
        """Initialize the async Starrocks adapter with a session manager.

        Args:
            orm_config: Starrocks-specific configuration. If None, uses global config.
        """
        configs = BaseConfig.global_config().STARROCKS_SQLALCHEMY if orm_config is None else orm_config
        super().__init__(configs)

    @override
    def _create_async_session_manager(
        self,
        configs: StarRocksSQLAlchemyConfig,
    ) -> AsyncStarRocksSQlAlchemySessionManager:
        """Create an async Starrocks-specific session manager.

        Args:
            configs: Starrocks configuration.

        Returns:
            An async Starrocks session manager instance.
        """
        return AsyncStarRocksSQlAlchemySessionManager(configs)

archipy.adapters.starrocks.sqlalchemy.adapters.AsyncStarrocksSQLAlchemyAdapter.session_manager instance-attribute 

session_manager: AsyncBaseSQLAlchemySessionManager[
    ConfigT
] = _create_async_session_manager(configs)

archipy.adapters.starrocks.sqlalchemy.adapters.AsyncStarrocksSQLAlchemyAdapter.get_session 

get_session() -> AsyncSession

Get a database session.

Returns:

Name Type Description
AsyncSession AsyncSession

A SQLAlchemy async session.

Raises:

Type Description
DatabaseConnectionError

If there's an error getting the session.
DatabaseConfigurationError

If there's an error in the database configuration.
Source code in archipy/adapters/base/sqlalchemy/adapters.py 
640
641
642
643
644
645
646
647
648
649
650
651
@override
def get_session(self) -> AsyncSession:
    """Get a database session.

    Returns:
        AsyncSession: A SQLAlchemy async session.

    Raises:
        DatabaseConnectionError: If there's an error getting the session.
        DatabaseConfigurationError: If there's an error in the database configuration.
    """
    return self.session_manager.get_session()

archipy.adapters.starrocks.sqlalchemy.adapters.AsyncStarrocksSQLAlchemyAdapter.execute_search_query async 

execute_search_query(
    entity: type[T],
    query: Select,
    pagination: PaginationDTO | None,
    sort_info: SortDTO | None = None,
    has_multiple_entities: bool = False,
) -> tuple[list[T], int]

Execute a search query with pagination and sorting.

Parameters:

Name Type Description Default
entity type[T]

The entity class to query.

 required
query Select

The SQLAlchemy SELECT query.

 required
pagination PaginationDTO | None

Optional pagination settings.

 required
sort_info SortDTO | None

Optional sorting information.

 None
has_multiple_entities bool

Optional bool

 False

Returns:

Type Description
tuple[list[T], int]

Tuple of the list of entities and the total count.

Raises:

Type Description
DatabaseQueryError

If the database query fails.
DatabaseTimeoutError

If the query times out.
DatabaseConnectionError

If there's a connection error.
DatabaseTransactionError

If there's a transaction error.
Source code in archipy/adapters/base/sqlalchemy/adapters.py 
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
@override
async def execute_search_query(
    self,
    entity: type[T],
    query: Select,
    pagination: PaginationDTO | None,
    sort_info: SortDTO | None = None,
    has_multiple_entities: bool = False,
) -> tuple[list[T], int]:
    """Execute a search query with pagination and sorting.

    Args:
        entity: The entity class to query.
        query: The SQLAlchemy SELECT query.
        pagination: Optional pagination settings.
        sort_info: Optional sorting information.
        has_multiple_entities: Optional bool

    Returns:
        Tuple of the list of entities and the total count.

    Raises:
        DatabaseQueryError: If the database query fails.
        DatabaseTimeoutError: If the query times out.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    try:
        sort_info = sort_info or SortDTO.default()
        session = self.get_session()
        sorted_query = self._apply_sorting(entity, query, sort_info)
        paginated_query = self._apply_pagination(sorted_query, pagination)
        result_set = await session.execute(paginated_query)
        if has_multiple_entities:
            # For multiple entities, fetchall returns list of Row objects
            raw_results = list(result_set.fetchall())
            # Convert to list[T] - each row contains entities of type T
            # Use tuple unpacking to access the first element
            results: list[T] = []
            for row in raw_results:
                if row:
                    # Row supports indexing and tuple unpacking
                    row_tuple = tuple(row)
                    if row_tuple:
                        first_entity = row_tuple[0]
                        # first_entity is T (entity type), verify it's an instance
                        if isinstance(first_entity, entity):
                            results.append(first_entity)
        else:
            # For single entity, scalars() returns list[T] directly
            results = list(result_set.scalars().all())
        count_query = select(func.count()).select_from(query.subquery())
        total_count_result = await session.execute(count_query)
        total_count = total_count_result.scalar_one()
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())
        raise  # This will never be reached, but satisfies MyPy
    else:
        # Type: results is list[T] where T extends BaseEntity, total_count is int
        return results, total_count

archipy.adapters.starrocks.sqlalchemy.adapters.AsyncStarrocksSQLAlchemyAdapter.create async 

create(entity: T) -> T | None

Create a new entity in the database.

Parameters:

Name Type Description Default
entity T

The entity to create.

 required

Returns:

Type Description
T | None

The created entity with updated attributes, preserving the original type.

Raises:

Type Description
InvalidEntityTypeError

If the entity type is not a valid SQLAlchemy model.
DatabaseQueryError

If the database operation fails.
DatabaseIntegrityError

If there's an integrity constraint violation.
DatabaseConstraintError

If there's a constraint violation.
DatabaseConnectionError

If there's a connection error.
DatabaseTransactionError

If there's a transaction error.
Source code in archipy/adapters/base/sqlalchemy/adapters.py 
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
@override
async def create(self, entity: T) -> T | None:
    """Create a new entity in the database.

    Args:
        entity: The entity to create.

    Returns:
        The created entity with updated attributes, preserving the original type.

    Raises:
        InvalidEntityTypeError: If the entity type is not a valid SQLAlchemy model.
        DatabaseQueryError: If the database operation fails.
        DatabaseIntegrityError: If there's an integrity constraint violation.
        DatabaseConstraintError: If there's a constraint violation.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    if not isinstance(entity, BaseEntity):
        raise InvalidEntityTypeError(
            message=f"Expected BaseEntity subclass, got {type(entity).__name__}",
            expected_type="BaseEntity",
            actual_type=type(entity).__name__,
        )

    try:
        session = self.get_session()
        session.add(entity)
        await session.flush()
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())
        raise  # This will never be reached, but satisfies MyPy
    else:
        return entity

archipy.adapters.starrocks.sqlalchemy.adapters.AsyncStarrocksSQLAlchemyAdapter.bulk_create async 

bulk_create(entities: list[T]) -> list[T] | None

Creates multiple entities in a single database operation.

Parameters:

Name Type Description Default
entities list[T]

List of entities to create.

 required

Returns:

Type Description
list[T] | None

List of created entities with updated attributes, preserving original types.

Raises:

Type Description
InvalidEntityTypeError

If any entity is not a valid SQLAlchemy model.
DatabaseQueryError

If the database operation fails.
DatabaseIntegrityError

If there's an integrity constraint violation.
DatabaseConstraintError

If there's a constraint violation.
DatabaseConnectionError

If there's a connection error.
DatabaseTransactionError

If there's a transaction error.
Source code in archipy/adapters/base/sqlalchemy/adapters.py 
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
@override
async def bulk_create(self, entities: list[T]) -> list[T] | None:
    """Creates multiple entities in a single database operation.

    Args:
        entities: List of entities to create.

    Returns:
        List of created entities with updated attributes, preserving original types.

    Raises:
        InvalidEntityTypeError: If any entity is not a valid SQLAlchemy model.
        DatabaseQueryError: If the database operation fails.
        DatabaseIntegrityError: If there's an integrity constraint violation.
        DatabaseConstraintError: If there's a constraint violation.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    if not all(isinstance(entity, BaseEntity) for entity in entities):
        raise InvalidEntityTypeError(
            message="All entities must be BaseEntity subclasses",
            expected_type="BaseEntity",
            actual_type="mixed",
        )

    try:
        session = self.get_session()
        session.add_all(entities)
        await session.flush()
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())
        raise  # This will never be reached, but satisfies MyPy
    else:
        return entities

archipy.adapters.starrocks.sqlalchemy.adapters.AsyncStarrocksSQLAlchemyAdapter.get_by_uuid async 

get_by_uuid(
    entity_type: type[T], entity_uuid: UUID
) -> BaseEntity | None

Retrieve an entity by its UUID.

Parameters:

Name Type Description Default
entity_type type[T]

The type of entity to retrieve.

 required
entity_uuid UUID

The UUID of the entity.

 required

Returns:

Type Description
BaseEntity | None

The entity if found, None otherwise.

Raises:

Type Description
InvalidEntityTypeError

If the entity type is not a valid SQLAlchemy model.
DatabaseQueryError

If the database operation fails.
DatabaseTimeoutError

If the query times out.
DatabaseConnectionError

If there's a connection error.
DatabaseTransactionError

If there's a transaction error.
Source code in archipy/adapters/base/sqlalchemy/adapters.py 
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
@override
async def get_by_uuid(self, entity_type: type[T], entity_uuid: UUID) -> BaseEntity | None:
    """Retrieve an entity by its UUID.

    Args:
        entity_type: The type of entity to retrieve.
        entity_uuid: The UUID of the entity.

    Returns:
        The entity if found, None otherwise.

    Raises:
        InvalidEntityTypeError: If the entity type is not a valid SQLAlchemy model.
        DatabaseQueryError: If the database operation fails.
        DatabaseTimeoutError: If the query times out.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    if not issubclass(entity_type, BaseEntity):
        raise InvalidEntityTypeError(
            message=f"Expected BaseEntity subclass, got {entity_type.__name__}",
            expected_type="BaseEntity",
            actual_type=entity_type.__name__,
        )

    try:
        session = self.get_session()
        result = await session.get(entity_type, entity_uuid)
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())
        raise  # This will never be reached, but satisfies MyPy
    else:
        # result is T | None where T extends BaseEntity, compatible with BaseEntity | None
        # The type checker needs explicit type annotation to understand the relationship
        typed_result: BaseEntity | None = result
        return typed_result

archipy.adapters.starrocks.sqlalchemy.adapters.AsyncStarrocksSQLAlchemyAdapter.delete async 

delete(entity: BaseEntity) -> None

Delete an entity from the database.

Parameters:

Name Type Description Default
entity BaseEntity

The entity to delete.

 required

Raises:

Type Description
InvalidEntityTypeError

If the entity is not a valid SQLAlchemy model.
DatabaseQueryError

If the database operation fails.
DatabaseIntegrityError

If there's an integrity constraint violation.
DatabaseConstraintError

If there's a constraint violation.
DatabaseConnectionError

If there's a connection error.
DatabaseTransactionError

If there's a transaction error.
Source code in archipy/adapters/base/sqlalchemy/adapters.py 
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
@override
async def delete(self, entity: BaseEntity) -> None:
    """Delete an entity from the database.

    Args:
        entity: The entity to delete.

    Raises:
        InvalidEntityTypeError: If the entity is not a valid SQLAlchemy model.
        DatabaseQueryError: If the database operation fails.
        DatabaseIntegrityError: If there's an integrity constraint violation.
        DatabaseConstraintError: If there's a constraint violation.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    if not isinstance(entity, BaseEntity):
        raise InvalidEntityTypeError(
            message=f"Expected BaseEntity subclass, got {type(entity).__name__}",
            expected_type="BaseEntity",
            actual_type=type(entity).__name__,
        )

    try:
        session = self.get_session()
        await session.delete(entity)
        await session.flush()
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())

archipy.adapters.starrocks.sqlalchemy.adapters.AsyncStarrocksSQLAlchemyAdapter.bulk_delete async 

bulk_delete(entities: list[T]) -> None

Delete multiple entities from the database.

Parameters:

Name Type Description Default
entities list[T]

List of entities to delete.

 required

Raises:

Type Description
InvalidEntityTypeError

If any entity is not a valid SQLAlchemy model.
DatabaseQueryError

If the database operation fails.
DatabaseIntegrityError

If there's an integrity constraint violation.
DatabaseConstraintError

If there's a constraint violation.
DatabaseConnectionError

If there's a connection error.
DatabaseTransactionError

If there's a transaction error.
Source code in archipy/adapters/base/sqlalchemy/adapters.py 
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
@override
async def bulk_delete(self, entities: list[T]) -> None:
    """Delete multiple entities from the database.

    Args:
        entities: List of entities to delete.

    Raises:
        InvalidEntityTypeError: If any entity is not a valid SQLAlchemy model.
        DatabaseQueryError: If the database operation fails.
        DatabaseIntegrityError: If there's an integrity constraint violation.
        DatabaseConstraintError: If there's a constraint violation.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    if not all(isinstance(entity, BaseEntity) for entity in entities):
        raise InvalidEntityTypeError(
            message="All entities must be BaseEntity subclasses",
            expected_type="BaseEntity",
            actual_type="mixed",
        )

    try:
        session = self.get_session()
        for entity in entities:
            await session.delete(entity)
        await session.flush()
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())

archipy.adapters.starrocks.sqlalchemy.adapters.AsyncStarrocksSQLAlchemyAdapter.execute async 

execute(
    statement: Executable,
    params: AnyExecuteParams | None = None,
) -> Result[Any]

Execute a SQLAlchemy statement.

Parameters:

Name Type Description Default
statement Executable

The SQLAlchemy statement to execute.

 required
params AnyExecuteParams | None

Optional parameters for the statement.

 None

Returns:

Type Description
Result[Any]

The result of the execution.

Raises:

Type Description
DatabaseQueryError

If the database operation fails.
DatabaseTimeoutError

If the query times out.
DatabaseConnectionError

If there's a connection error.
DatabaseTransactionError

If there's a transaction error.
Source code in archipy/adapters/base/sqlalchemy/adapters.py 
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
@override
async def execute(self, statement: Executable, params: AnyExecuteParams | None = None) -> Result[Any]:
    """Execute a SQLAlchemy statement.

    Args:
        statement: The SQLAlchemy statement to execute.
        params: Optional parameters for the statement.

    Returns:
        The result of the execution.

    Raises:
        DatabaseQueryError: If the database operation fails.
        DatabaseTimeoutError: If the query times out.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    try:
        session = self.get_session()
        result = await session.execute(statement, params or {})
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())
        raise  # This will never be reached, but satisfies MyPy
    else:
        return result

archipy.adapters.starrocks.sqlalchemy.adapters.AsyncStarrocksSQLAlchemyAdapter.scalars async 

scalars(
    statement: Executable,
    params: AnyExecuteParams | None = None,
) -> ScalarResult[Any]

Execute a SQLAlchemy statement and return scalar results.

Parameters:

Name Type Description Default
statement Executable

The SQLAlchemy statement to execute.

 required
params AnyExecuteParams | None

Optional parameters for the statement.

 None

Returns:

Type Description
ScalarResult[Any]

The scalar results of the execution.

Raises:

Type Description
DatabaseQueryError

If the database operation fails.
DatabaseTimeoutError

If the query times out.
DatabaseConnectionError

If there's a connection error.
DatabaseTransactionError

If there's a transaction error.
Source code in archipy/adapters/base/sqlalchemy/adapters.py 
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
@override
async def scalars(self, statement: Executable, params: AnyExecuteParams | None = None) -> ScalarResult[Any]:
    """Execute a SQLAlchemy statement and return scalar results.

    Args:
        statement: The SQLAlchemy statement to execute.
        params: Optional parameters for the statement.

    Returns:
        The scalar results of the execution.

    Raises:
        DatabaseQueryError: If the database operation fails.
        DatabaseTimeoutError: If the query times out.
        DatabaseConnectionError: If there's a connection error.
        DatabaseTransactionError: If there's a transaction error.
    """
    try:
        session = self.get_session()
        result = await session.scalars(statement, params or {})
    except Exception as e:
        self._handle_db_exception(e, self.session_manager._get_database_name())
        raise  # This will never be reached, but satisfies MyPy
    else:
        return result

options: show_root_toc_entry: false heading_level: 3