Create an async SQLAlchemy engine from a database URL. Args: url: Database connection URL. Examples: - postgresql+asyncpg://user:pass@localhost:5432/orgkernel - mysql+aiomysql://user:pass@localhost:3306/orgkernel For S
(url: str, **kwargs: Any)
| 42 | |
| 43 | |
| 44 | def get_engine(url: str, **kwargs: Any) -> AsyncEngine: |
| 45 | """ |
| 46 | Create an async SQLAlchemy engine from a database URL. |
| 47 | |
| 48 | Args: |
| 49 | url: Database connection URL. |
| 50 | Examples: |
| 51 | - postgresql+asyncpg://user:pass@localhost:5432/orgkernel |
| 52 | - mysql+aiomysql://user:pass@localhost:3306/orgkernel |
| 53 | For SQLite (dev/testing): sqlite+aiosqlite:///./orgkernel.db |
| 54 | **kwargs: Additional arguments passed to create_async_engine. |
| 55 | |
| 56 | Returns: |
| 57 | Configured AsyncEngine instance. |
| 58 | """ |
| 59 | return create_async_engine(url, echo=False, **kwargs) |
| 60 | |
| 61 | |
| 62 | def get_session_factory(engine: AsyncEngine) -> type[AsyncSession]: |
nothing calls this directly
no outgoing calls
no test coverage detected