Source code for codegen_database.ext.pg_cron

from __future__ import annotations

from dataclasses import dataclass
from typing import TYPE_CHECKING

from codegen_database.ext.cron.alembic import register_cron_alembic_events
from codegen_database.ext.cron.base import CronJob, register_cron_job
from codegen_database.extension import CodegenDatabaseExtension
from codegen_database.pg_extension import (
    PGExtension,
    register_pg_extension,
)

if TYPE_CHECKING:
    from sqlalchemy import MetaData

__all__ = [
    "CronJob",
    "PGCronExtension",
    "PGExtension",
    "register_cron_job",
    "register_pg_extension",
]

_PG_CRON_EXTENSION = PGExtension("pg_cron", schema="cron", cascade=True)




[docs]
@dataclass(frozen=True, eq=True)
class PGCronExtension(CodegenDatabaseExtension):
    name: str = "pg_cron"



[docs]
    def configure_metadata(self, metadata: MetaData) -> None:
        register_pg_extension(metadata, _PG_CRON_EXTENSION)





[docs]
    def configure_alembic(self) -> None:
        register_cron_alembic_events()