this is the schedule manager not the db
This commit is contained in:
C committed
1 parent
f0e1d79e92
commit
0f60a31ef8
1 file changed
+221
-221
@@ -1,221 +1,221 @@
|
|||||||
import aiosqlite
|
import aiosqlite
|
||||||
import logging
|
import logging
|
||||||
from typing import Optional, List, Dict, Any
|
from typing import Optional, List, Dict, Any
|
||||||
import json
|
import json
|
||||||
import os
|
import os
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
# Get the absolute path to edderkop.db in the project root
|
# Get the absolute path to edderkop.db in the project root
|
||||||
DEFAULT_DB_PATH = os.path.join(
|
DEFAULT_DB_PATH = os.path.join(
|
||||||
os.path.dirname(os.path.dirname(__file__)), "edderkop.db"
|
os.path.dirname(os.path.dirname(__file__)), "edderkop.db"
|
||||||
)
|
)
|
||||||
|
|
||||||
if not os.path.exists(DEFAULT_DB_PATH):
|
if not os.path.exists(DEFAULT_DB_PATH):
|
||||||
raise FileNotFoundError(f"Database file not found at {DEFAULT_DB_PATH}")
|
raise FileNotFoundError(f"Database file not found at {DEFAULT_DB_PATH}")
|
||||||
print('Please run "src/scripts/EDDERKOP.SQL" to setup the database.')
|
print('Please run "src/scripts/EDDERKOP.SQL" to setup the database.')
|
||||||
os.exit(0)
|
os.exit(0)
|
||||||
|
|
||||||
|
|
||||||
class AsyncSchedulerDB:
|
class AsyncSchedulerDB:
|
||||||
"""
|
"""
|
||||||
Asynchronous SQLite database manager for APScheduler.
|
Asynchronous SQLite database manager for APScheduler.
|
||||||
Handles job storage, retrieval, and database operations using the main edderkop.db database.
|
Handles job storage, retrieval, and database operations using the main edderkop.db database.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, db_path: str = DEFAULT_DB_PATH):
|
def __init__(self, db_path: str = DEFAULT_DB_PATH):
|
||||||
"""
|
"""
|
||||||
Initialize the database manager.
|
Initialize the database manager.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
db_path (str, optional): Path to the SQLite database file.
|
db_path (str, optional): Path to the SQLite database file.
|
||||||
Defaults to edderkop.db in project root.
|
Defaults to edderkop.db in project root.
|
||||||
"""
|
"""
|
||||||
self.db_path = db_path
|
self.db_path = db_path
|
||||||
self._initialized = False
|
self._initialized = False
|
||||||
logger.info(f"Using database at: {self.db_path}")
|
logger.info(f"Using database at: {self.db_path}")
|
||||||
|
|
||||||
async def initialize(self):
|
async def initialize(self):
|
||||||
"""Initialize the database schema if it doesn't exist."""
|
"""Initialize the database schema if it doesn't exist."""
|
||||||
if self._initialized:
|
if self._initialized:
|
||||||
return
|
return
|
||||||
|
|
||||||
async with aiosqlite.connect(self.db_path) as db:
|
async with aiosqlite.connect(self.db_path) as db:
|
||||||
# Create apscheduler jobs table if it doesn't exist
|
# Create apscheduler jobs table if it doesn't exist
|
||||||
await db.execute(
|
await db.execute(
|
||||||
"""
|
"""
|
||||||
CREATE TABLE IF NOT EXISTS apscheduler_jobs (
|
CREATE TABLE IF NOT EXISTS apscheduler_jobs (
|
||||||
id VARCHAR(191) NOT NULL,
|
id VARCHAR(191) NOT NULL,
|
||||||
next_run_time FLOAT,
|
next_run_time FLOAT,
|
||||||
job_state BLOB NOT NULL,
|
job_state BLOB NOT NULL,
|
||||||
PRIMARY KEY (id)
|
PRIMARY KEY (id)
|
||||||
)
|
)
|
||||||
"""
|
"""
|
||||||
)
|
)
|
||||||
await db.commit()
|
await db.commit()
|
||||||
|
|
||||||
self._initialized = True
|
self._initialized = True
|
||||||
logger.info(f"Initialized scheduler database at {self.db_path}")
|
logger.info(f"Initialized scheduler database at {self.db_path}")
|
||||||
|
|
||||||
async def get_all_jobs(self) -> List[Dict[str, Any]]:
|
async def get_all_jobs(self) -> List[Dict[str, Any]]:
|
||||||
"""
|
"""
|
||||||
Retrieve all jobs from the database.
|
Retrieve all jobs from the database.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
List[Dict[str, Any]]: List of job dictionaries
|
List[Dict[str, Any]]: List of job dictionaries
|
||||||
"""
|
"""
|
||||||
await self.initialize()
|
await self.initialize()
|
||||||
async with aiosqlite.connect(self.db_path) as db:
|
async with aiosqlite.connect(self.db_path) as db:
|
||||||
db.row_factory = aiosqlite.Row
|
db.row_factory = aiosqlite.Row
|
||||||
async with db.execute(
|
async with db.execute(
|
||||||
"SELECT * FROM apscheduler_jobs ORDER BY next_run_time"
|
"SELECT * FROM apscheduler_jobs ORDER BY next_run_time"
|
||||||
) as cursor:
|
) as cursor:
|
||||||
rows = await cursor.fetchall()
|
rows = await cursor.fetchall()
|
||||||
return [
|
return [
|
||||||
{
|
{
|
||||||
"id": row["id"],
|
"id": row["id"],
|
||||||
"next_run_time": row["next_run_time"],
|
"next_run_time": row["next_run_time"],
|
||||||
"job_state": json.loads(row["job_state"]),
|
"job_state": json.loads(row["job_state"]),
|
||||||
}
|
}
|
||||||
for row in rows
|
for row in rows
|
||||||
]
|
]
|
||||||
|
|
||||||
async def get_job(self, job_id: str) -> Optional[Dict[str, Any]]:
|
async def get_job(self, job_id: str) -> Optional[Dict[str, Any]]:
|
||||||
"""
|
"""
|
||||||
Retrieve a specific job by ID.
|
Retrieve a specific job by ID.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
job_id (str): The ID of the job to retrieve
|
job_id (str): The ID of the job to retrieve
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
Optional[Dict[str, Any]]: Job dictionary if found, None otherwise
|
Optional[Dict[str, Any]]: Job dictionary if found, None otherwise
|
||||||
"""
|
"""
|
||||||
await self.initialize()
|
await self.initialize()
|
||||||
async with aiosqlite.connect(self.db_path) as db:
|
async with aiosqlite.connect(self.db_path) as db:
|
||||||
db.row_factory = aiosqlite.Row
|
db.row_factory = aiosqlite.Row
|
||||||
async with db.execute(
|
async with db.execute(
|
||||||
"SELECT * FROM apscheduler_jobs WHERE id = ?", (job_id,)
|
"SELECT * FROM apscheduler_jobs WHERE id = ?", (job_id,)
|
||||||
) as cursor:
|
) as cursor:
|
||||||
row = await cursor.fetchone()
|
row = await cursor.fetchone()
|
||||||
if row:
|
if row:
|
||||||
return {
|
return {
|
||||||
"id": row["id"],
|
"id": row["id"],
|
||||||
"next_run_time": row["next_run_time"],
|
"next_run_time": row["next_run_time"],
|
||||||
"job_state": json.loads(row["job_state"]),
|
"job_state": json.loads(row["job_state"]),
|
||||||
}
|
}
|
||||||
return None
|
return None
|
||||||
|
|
||||||
async def add_job(
|
async def add_job(
|
||||||
self, job_id: str, next_run_time: float, job_state: Dict[str, Any]
|
self, job_id: str, next_run_time: float, job_state: Dict[str, Any]
|
||||||
) -> bool:
|
) -> bool:
|
||||||
"""
|
"""
|
||||||
Add a new job to the database.
|
Add a new job to the database.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
job_id (str): Unique identifier for the job
|
job_id (str): Unique identifier for the job
|
||||||
next_run_time (float): Unix timestamp for next scheduled run
|
next_run_time (float): Unix timestamp for next scheduled run
|
||||||
job_state (Dict[str, Any]): Job state dictionary
|
job_state (Dict[str, Any]): Job state dictionary
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
bool: True if successful, False otherwise
|
bool: True if successful, False otherwise
|
||||||
"""
|
"""
|
||||||
await self.initialize()
|
await self.initialize()
|
||||||
try:
|
try:
|
||||||
async with aiosqlite.connect(self.db_path) as db:
|
async with aiosqlite.connect(self.db_path) as db:
|
||||||
await db.execute(
|
await db.execute(
|
||||||
"INSERT INTO apscheduler_jobs (id, next_run_time, job_state) VALUES (?, ?, ?)",
|
"INSERT INTO apscheduler_jobs (id, next_run_time, job_state) VALUES (?, ?, ?)",
|
||||||
(job_id, next_run_time, json.dumps(job_state)),
|
(job_id, next_run_time, json.dumps(job_state)),
|
||||||
)
|
)
|
||||||
await db.commit()
|
await db.commit()
|
||||||
return True
|
return True
|
||||||
except aiosqlite.IntegrityError:
|
except aiosqlite.IntegrityError:
|
||||||
logger.error(f"Job with ID {job_id} already exists")
|
logger.error(f"Job with ID {job_id} already exists")
|
||||||
return False
|
return False
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error(f"Error adding job: {e}")
|
logger.error(f"Error adding job: {e}")
|
||||||
return False
|
return False
|
||||||
|
|
||||||
async def update_job(
|
async def update_job(
|
||||||
self, job_id: str, next_run_time: float, job_state: Dict[str, Any]
|
self, job_id: str, next_run_time: float, job_state: Dict[str, Any]
|
||||||
) -> bool:
|
) -> bool:
|
||||||
"""
|
"""
|
||||||
Update an existing job in the database.
|
Update an existing job in the database.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
job_id (str): ID of the job to update
|
job_id (str): ID of the job to update
|
||||||
next_run_time (float): New next run time
|
next_run_time (float): New next run time
|
||||||
job_state (Dict[str, Any]): Updated job state
|
job_state (Dict[str, Any]): Updated job state
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
bool: True if successful, False if job not found
|
bool: True if successful, False if job not found
|
||||||
"""
|
"""
|
||||||
await self.initialize()
|
await self.initialize()
|
||||||
async with aiosqlite.connect(self.db_path) as db:
|
async with aiosqlite.connect(self.db_path) as db:
|
||||||
cursor = await db.execute(
|
cursor = await db.execute(
|
||||||
"""UPDATE apscheduler_jobs
|
"""UPDATE apscheduler_jobs
|
||||||
SET next_run_time = ?, job_state = ?
|
SET next_run_time = ?, job_state = ?
|
||||||
WHERE id = ?""",
|
WHERE id = ?""",
|
||||||
(next_run_time, json.dumps(job_state), job_id),
|
(next_run_time, json.dumps(job_state), job_id),
|
||||||
)
|
)
|
||||||
await db.commit()
|
await db.commit()
|
||||||
return cursor.rowcount > 0
|
return cursor.rowcount > 0
|
||||||
|
|
||||||
async def remove_job(self, job_id: str) -> bool:
|
async def remove_job(self, job_id: str) -> bool:
|
||||||
"""
|
"""
|
||||||
Remove a job from the database.
|
Remove a job from the database.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
job_id (str): ID of the job to remove
|
job_id (str): ID of the job to remove
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
bool: True if job was removed, False if not found
|
bool: True if job was removed, False if not found
|
||||||
"""
|
"""
|
||||||
await self.initialize()
|
await self.initialize()
|
||||||
async with aiosqlite.connect(self.db_path) as db:
|
async with aiosqlite.connect(self.db_path) as db:
|
||||||
cursor = await db.execute(
|
cursor = await db.execute(
|
||||||
"DELETE FROM apscheduler_jobs WHERE id = ?", (job_id,)
|
"DELETE FROM apscheduler_jobs WHERE id = ?", (job_id,)
|
||||||
)
|
)
|
||||||
await db.commit()
|
await db.commit()
|
||||||
return cursor.rowcount > 0
|
return cursor.rowcount > 0
|
||||||
|
|
||||||
async def remove_all_jobs(self) -> int:
|
async def remove_all_jobs(self) -> int:
|
||||||
"""
|
"""
|
||||||
Remove all jobs from the database.
|
Remove all jobs from the database.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
int: Number of jobs removed
|
int: Number of jobs removed
|
||||||
"""
|
"""
|
||||||
await self.initialize()
|
await self.initialize()
|
||||||
async with aiosqlite.connect(self.db_path) as db:
|
async with aiosqlite.connect(self.db_path) as db:
|
||||||
cursor = await db.execute("DELETE FROM apscheduler_jobs")
|
cursor = await db.execute("DELETE FROM apscheduler_jobs")
|
||||||
await db.commit()
|
await db.commit()
|
||||||
return cursor.rowcount
|
return cursor.rowcount
|
||||||
|
|
||||||
async def get_due_jobs(self, now: float) -> List[Dict[str, Any]]:
|
async def get_due_jobs(self, now: float) -> List[Dict[str, Any]]:
|
||||||
"""
|
"""
|
||||||
Get all jobs that are due to run.
|
Get all jobs that are due to run.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
now (float): Current timestamp
|
now (float): Current timestamp
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
List[Dict[str, Any]]: List of due jobs
|
List[Dict[str, Any]]: List of due jobs
|
||||||
"""
|
"""
|
||||||
await self.initialize()
|
await self.initialize()
|
||||||
async with aiosqlite.connect(self.db_path) as db:
|
async with aiosqlite.connect(self.db_path) as db:
|
||||||
db.row_factory = aiosqlite.Row
|
db.row_factory = aiosqlite.Row
|
||||||
async with db.execute(
|
async with db.execute(
|
||||||
"""SELECT * FROM apscheduler_jobs
|
"""SELECT * FROM apscheduler_jobs
|
||||||
WHERE next_run_time <= ?
|
WHERE next_run_time <= ?
|
||||||
ORDER BY next_run_time""",
|
ORDER BY next_run_time""",
|
||||||
(now,),
|
(now,),
|
||||||
) as cursor:
|
) as cursor:
|
||||||
rows = await cursor.fetchall()
|
rows = await cursor.fetchall()
|
||||||
return [
|
return [
|
||||||
{
|
{
|
||||||
"id": row["id"],
|
"id": row["id"],
|
||||||
"next_run_time": row["next_run_time"],
|
"next_run_time": row["next_run_time"],
|
||||||
"job_state": json.loads(row["job_state"]),
|
"job_state": json.loads(row["job_state"]),
|
||||||
}
|
}
|
||||||
for row in rows
|
for row in rows
|
||||||
]
|
]
|
||||||
Reference in new issue
Block a user