Skip to content

rs_dpr_service/jobs_table.md

<< Back to index

Module used to implement abstract model of an SQLAlchemy table.

JobType

Bases: Enum

Enum for the job type specified in OGC API Processes specification

Source code in docs/rs-dpr-service/rs_dpr_service/jobs_table.py 
34
35
36
37
38
39
40
class JobType(enum.Enum):
    """
    Enum for the job type specified in OGC API Processes specification
    """

    #  From the specification
    process = "process"  # pylint: disable=invalid-name

JobsTable

Bases: Base

Abstract implementation of SQLAlchemy Base

Must be kept in line with https://github.com/geopython/pygeoapi/blob/master/tests/data/postgres_manager_full_structure.backup.sql

Source code in docs/rs-dpr-service/rs_dpr_service/jobs_table.py 
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
class JobsTable(Base):  # pylint: disable=too-few-public-methods
    """
    Abstract implementation of SQLAlchemy Base

    Must be kept in line with
    https://github.com/geopython/pygeoapi/blob/master/tests/data/postgres_manager_full_structure.backup.sql
    """

    __tablename__ = "jobs"

    type = Column(Enum(JobType), nullable=False, server_default=JobType.process.value)
    identifier = Column(String, primary_key=True, unique=True, index=True)
    processID = Column(String, nullable=False)
    status = Column(Enum(JobStatus), nullable=False)
    progress = Column(Integer, server_default="0", nullable=False)
    # Pylint issue with func.now, check this: https://github.com/sqlalchemy/sqlalchemy/issues/9189
    created = Column(DateTime, server_default=func.now())  # pylint: disable=not-callable
    started = Column(DateTime, server_default=func.now())  # pylint: disable=not-callable
    finished = Column(DateTime)  # pylint: disable=not-callable
    # onupdate=func.now(), server_onupdate=func.now() is not working, did not figure why
    # instead, force the PostgreSQLManager from pygeoapi to update the updated column specifically with
    # update_job function (check processors.py log_job_execution function)
    updated = Column(
        DateTime,
        server_default=func.now(),  # pylint: disable=not-callable
        onupdate=func.now(),  # pylint: disable=not-callable
        server_onupdate=func.now(),  # pylint: disable=not-callable
    )
    location = Column(String)
    mimetype = Column(String)
    message = Column(String)

    def __init__(self, *args, **kwargs):
        """Invoked when creating a new record in the database table."""
        super().__init__(*args, **kwargs)
        self.lock = Lock()

    @orm.reconstructor
    def init_on_load(self):
        """Invoked when retrieving an existing record from the database table."""
        self.lock = Lock()

__init__(*args, **kwargs)

Invoked when creating a new record in the database table.

Source code in docs/rs-dpr-service/rs_dpr_service/jobs_table.py 
75
76
77
78
def __init__(self, *args, **kwargs):
    """Invoked when creating a new record in the database table."""
    super().__init__(*args, **kwargs)
    self.lock = Lock()

init_on_load()

Invoked when retrieving an existing record from the database table.

Source code in docs/rs-dpr-service/rs_dpr_service/jobs_table.py 
80
81
82
83
@orm.reconstructor
def init_on_load(self):
    """Invoked when retrieving an existing record from the database table."""
    self.lock = Lock()