Use CrateDB as JobStore #7
Merged
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
amotl
force-pushed
the
cratedb-storage
branch
2 times, most recently
from
300a04a
to
ee50caa
Compare
amotl
force-pushed
the
cratedb-storage
branch
2 times, most recently
from
c30fb9c
to
dc6a828
Compare
amotl
commented
Dec 2, 2023
Comment on lines
+1
to
+26
| """ | ||
| class CrateDBMongoDBJobStore(MongoDBJobStore): | ||
| from cratedb_toolkit.adapter.pymongo import PyMongoCrateDbAdapter | ||
| def __init__(self, dburi, *args, **kwargs): | ||
| with PyMongoCrateDbAdapter(dburi=dburi): | ||
| super().__init__(*args, **kwargs) | ||
| """ | ||
|
|
||
|
|
||
| def CrateDBMongoDBJobStore(dburi: str): | ||
| from cratedb_toolkit.adapter.pymongo import PyMongoCrateDbAdapter | ||
|
|
||
| with PyMongoCrateDbAdapter(dburi=dburi): | ||
| import pymongo | ||
|
|
||
| amalgamated_client: pymongo.MongoClient = pymongo.MongoClient( | ||
| "localhost", 27017, timeoutMS=100, connectTimeoutMS=100, socketTimeoutMS=100, serverSelectionTimeoutMS=100 | ||
| ) | ||
| from apscheduler.jobstores.mongodb import MongoDBJobStore | ||
|
|
||
| class CrateDBMongoDBJobStoreImpl(MongoDBJobStore): | ||
| def __init__(self, *args, **kwargs): | ||
| kwargs["client"] = amalgamated_client | ||
| super().__init__(*args, **kwargs) | ||
|
|
||
| return CrateDBMongoDBJobStoreImpl() |
There was a problem hiding this comment.
It was just an attempt, and will be removed.
Comment on lines
+7
to
+15
| class LargeBinary(sa.String): | ||
|
|
||
| """A type for large binary byte data. | ||
|
|
||
| The :class:`.LargeBinary` type corresponds to a large and/or unlengthed | ||
| binary type for the target platform, such as BLOB on MySQL and BYTEA for | ||
| PostgreSQL. It also handles the necessary conversions for the DBAPI. | ||
|
|
||
| """ |
There was a problem hiding this comment.
This needs to go into crate-python.
Comment on lines
+67
to
+112
| def patchme(self): | ||
| """ | ||
| A few patches to make the CrateDB SQLAlchemy dialect work. | ||
|
|
||
| TODO: Upstream to crate-python. | ||
| """ | ||
| from crate.client.sqlalchemy import CrateDialect | ||
| from crate.client.sqlalchemy.compiler import CrateDDLCompiler, CrateTypeCompiler | ||
|
|
||
| def visit_BLOB(self, type_, **kw): | ||
| return "STRING" | ||
|
|
||
| def visit_FLOAT(self, type_, **kw): | ||
| """ | ||
| From `sqlalchemy.sql.sqltypes.Float`. | ||
|
|
||
| When a :paramref:`.Float.precision` is not provided in a | ||
| :class:`_types.Float` type some backend may compile this type as | ||
| an 8 bytes / 64 bit float datatype. To use a 4 bytes / 32 bit float | ||
| datatype a precision <= 24 can usually be provided or the | ||
| :class:`_types.REAL` type can be used. | ||
| This is known to be the case in the PostgreSQL and MSSQL dialects | ||
| that render the type as ``FLOAT`` that's in both an alias of | ||
| ``DOUBLE PRECISION``. Other third party dialects may have similar | ||
| behavior. | ||
| """ | ||
| if not type_.precision: | ||
| return "FLOAT" | ||
| elif type_.precision <= 24: | ||
| return "FLOAT" | ||
| else: | ||
| return "DOUBLE" | ||
|
|
||
| CrateTypeCompiler.visit_BLOB = visit_BLOB | ||
| CrateTypeCompiler.visit_FLOAT = visit_FLOAT | ||
|
|
||
| def visit_create_index(self, create, **kw): | ||
| return "SELECT 1;" | ||
|
|
||
| CrateDDLCompiler.visit_create_index = visit_create_index | ||
|
|
||
| CrateDialect.colspecs.update( | ||
| { | ||
| sa.sql.sqltypes.LargeBinary: LargeBinary, | ||
| } | ||
| ) |
There was a problem hiding this comment.
This also needs to go into crate-python.
Comment on lines
+56
to
+65
| def receive_after_execute( | ||
| conn: sa.engine.Connection, clauseelement, multiparams, params, execution_options, result | ||
| ): | ||
| """ | ||
| Run a `REFRESH TABLE ...` command after each DML operation (INSERT, UPDATE, DELETE). | ||
| """ | ||
| if isinstance(clauseelement, (sa.sql.Insert, sa.sql.Update, sa.sql.Delete)): | ||
| conn.execute(sa.text(f"REFRESH TABLE {clauseelement.table}")) | ||
|
|
||
| sa.event.listen(self.engine, "after_execute", receive_after_execute) |
There was a problem hiding this comment.
On the other hand, this needs to go into cratedb-toolkit.
amotl
force-pushed
the
cratedb-storage
branch
from
dc6a828
to
d141f83
Compare
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
About
APScheduler provides a few built-in job store implementations already. This patch builds upon the
SQLAlchemyJobStore, and makes it compatible with CrateDB by adding a few polyfills. 1Footnotes
As far as I can see, most of those can be refactored into the SQLAlchemy dialect in the canonical
crate-pythonpackage. ↩