mirror of
https://github.com/hanzo-ds/connect.git
synced 2026-10-30 20:31:24 +00:00
Joe SpadolaandGitHub
7ce2499f68
* sqa dict and alembic work * sqa dialect improvements * import shim * engine shim * implement alembic support * clickhouse dictionary support * compat shim * add alembic to test reqs * integration/unit tests * update readme * update changelog * fix to support sqa 1.4 and 2.x * update readme * add alembic worked example * consolidate duplicate code * add sql parse logic file * cloud quirk reflection fixes * cc_sqalchemy ergonomics * chcsqa prewhere, limit by, lambda chainables * preserve ARRAY JOIN labels, fix Tuple adapt * wording edits, name fixes * linting * linting, use format * simplify docstrings * update migration guide * update migration guide * fix alembic enum render for autogen migrations * fix alembic enum render for nested container types * 1.1.0a1 release prep * preserve CH select modifiers when compiling from ORM-mapped attributes * 1.1.0a2 release prep * address pr comments
54 lines
2.0 KiB
Python
54 lines
2.0 KiB
Python
from sqlalchemy import Table
|
|
|
|
_DICTIONARY_KWARGS = ("source", "layout", "lifetime", "primary_key")
|
|
|
|
|
|
def _pop_dictionary_kwargs(kwargs):
|
|
"""Pop Dictionary-specific kwargs before Table validates them."""
|
|
return {k: kwargs.pop(k, None) for k in _DICTIONARY_KWARGS}
|
|
|
|
|
|
def _apply_dictionary_metadata(table, popped):
|
|
"""Set dialect-prefixed kwargs on the table after construction."""
|
|
table.kwargs["clickhouse_table_type"] = "dictionary"
|
|
if popped.get("source") is not None:
|
|
table.source = popped["source"]
|
|
table.kwargs["clickhouse_dictionary_source"] = popped["source"]
|
|
if popped.get("layout") is not None:
|
|
table.layout = popped["layout"]
|
|
table.kwargs["clickhouse_dictionary_layout"] = popped["layout"]
|
|
if popped.get("lifetime") is not None:
|
|
table.lifetime = popped["lifetime"]
|
|
table.kwargs["clickhouse_dictionary_lifetime"] = popped["lifetime"]
|
|
if popped.get("primary_key") is not None:
|
|
table.primary_key_def = popped["primary_key"]
|
|
table.kwargs["clickhouse_dictionary_primary_key"] = popped["primary_key"]
|
|
|
|
|
|
class Dictionary(Table):
|
|
"""
|
|
Represents a ClickHouse Dictionary.
|
|
|
|
Inherits from Table so it can be attached to metadata and have columns.
|
|
|
|
Custom kwargs must be intercepted before Table's dialect-kwarg validation
|
|
runs. The interception point differs between SQLAlchemy versions:
|
|
|
|
- SQA 1.4: Table.__new__ calls _init() directly, bypassing __init__
|
|
- SQA 2.x: Table.__new__ calls __init__() directly (no _init)
|
|
|
|
We override both to handle either path.
|
|
"""
|
|
|
|
__visit_name__ = "dictionary"
|
|
|
|
def __init__(self, name, metadata, *args, **kwargs):
|
|
popped = _pop_dictionary_kwargs(kwargs)
|
|
super().__init__(name, metadata, *args, **kwargs)
|
|
_apply_dictionary_metadata(self, popped)
|
|
|
|
def _init(self, name, metadata, *args, **kwargs):
|
|
popped = _pop_dictionary_kwargs(kwargs)
|
|
super()._init(name, metadata, *args, **kwargs)
|
|
_apply_dictionary_metadata(self, popped)
|