Add create shcmea for sqlalchemy datastore

docs/versionhistory.rst

@@ -60,6 +60,8 @@ APScheduler, see the :doc:`migration section <migration>`.
   useful and reasonable
 - Fixed race condition in ``MongoDBDataStore`` that allowed multiple schedulers to
   acquire the same schedules at once
+- Changed ``SQLAlchemyDataStore`` to automatically create the explicitly specified
+  schema if it's missing
 
 **4.0.0a5**
 

src/apscheduler/datastores/sqlalchemy.py

@@ -48,6 +48,7 @@
 )
 from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine, create_async_engine
 from sqlalchemy.future import Connection, Engine
+from sqlalchemy.schema import CreateSchema
 from sqlalchemy.sql import Executable
 from sqlalchemy.sql.ddl import DropTable
 from sqlalchemy.sql.elements import BindParameter, literal
@@ -231,6 +232,12 @@ async def _begin_transaction(
 
             yield conn
 
+    async def _create_schema(self, conn: Connection | AsyncConnection) -> None:
+        if not self.schema:
+            return
+        t = CreateSchema(name=self.schema, if_not_exists=True)
+        await self._execute(conn, t)
+
     async def _create_metadata(self, conn: Connection | AsyncConnection) -> None:
         if isinstance(conn, AsyncConnection):
             await conn.run_sync(self._metadata.create_all)
@@ -389,6 +396,7 @@ async def start(
         async for attempt in self._retry():
             with attempt:
                 async with self._begin_transaction() as conn:
+                    await self._create_schema(conn)
                     if self.start_from_scratch:
                         for table in self._metadata.sorted_tables:
                             await self._execute(conn, DropTable(table, if_exists=True))