#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
"""RedisHook module."""
from __future__ import annotations
import inspect
from typing import Any
import redis
from redis import Redis
from airflow.providers.common.compat.sdk import BaseHook
from airflow.providers.redis import __version__ as provider_version
[docs]
DriverInfo = getattr(redis, "DriverInfo", None)
[docs]
DEFAULT_SSL_CERT_REQS = "required"
[docs]
ALLOWED_SSL_CERT_REQS = [DEFAULT_SSL_CERT_REQS, "optional", "none"]
# Check at module import time what Redis client identification features are supported
_REDIS_PARAMS = inspect.signature(Redis.__init__).parameters
_SUPPORTS_LIB_NAME = "lib_name" in _REDIS_PARAMS
[docs]
class RedisHook(BaseHook):
"""
Wrapper for connection to interact with Redis in-memory data structure store.
You can set your db in the extra field of your connection as ``{"db": 3}``.
Also you can set ssl parameters as:
``{"ssl": true, "ssl_cert_reqs": "require", "ssl_certfile": "/path/to/cert.pem", etc}``.
"""
[docs]
conn_name_attr = "redis_conn_id"
[docs]
default_conn_name = "redis_default"
def __init__(self, redis_conn_id: str = default_conn_name, **kwargs) -> None:
"""
Prepare hook to connect to a Redis database.
:param conn_id: the name of the connection that has the parameters
we need to connect to Redis.
"""
super().__init__()
[docs]
self.redis_conn_id = redis_conn_id
[docs]
self.host = kwargs.get("host", None)
[docs]
self.port = kwargs.get("port", None)
[docs]
self.username = kwargs.get("username", None)
[docs]
self.password = kwargs.get("password", None)
[docs]
self.db = kwargs.get("db", None)
[docs]
def get_conn(self):
"""Return a Redis connection."""
conn = self.get_connection(self.redis_conn_id)
self.host = conn.host
self.port = conn.port
self.username = conn.login
self.password = None if str(conn.password).lower() in ["none", "false", ""] else conn.password
self.db = conn.extra_dejson.get("db")
# check for ssl parameters in conn.extra
ssl_arg_names = [
"ssl",
"ssl_cert_reqs",
"ssl_ca_certs",
"ssl_keyfile",
"ssl_certfile",
"ssl_check_hostname",
]
ssl_args = {name: val for name, val in conn.extra_dejson.items() if name in ssl_arg_names}
if not self.redis:
self.log.debug(
'Initializing redis object for conn_id "%s" on %s:%s:%s',
self.redis_conn_id,
self.host,
self.port,
self.db,
)
# Add driver info for client identification if supported
# This allows Redis server to identify the Redis provider as the upstream driver.
# See: https://redis.io/docs/latest/commands/client-setinfo/
driver_info_options: dict[str, Any] = {}
if DriverInfo is not None:
driver_info = DriverInfo().add_upstream_driver(
"apache-airflow-providers-redis", provider_version
)
driver_info_options = {"driver_info": driver_info}
elif _SUPPORTS_LIB_NAME:
driver_info_options = {
"lib_name": f"redis-py(apache-airflow-providers-redis_v{provider_version})",
}
self.redis = Redis(
host=self.host,
port=self.port,
username=self.username,
password=self.password,
db=self.db,
**ssl_args,
**driver_info_options,
)
return self.redis
@classmethod
[docs]
def get_ui_field_behaviour(cls) -> dict[str, Any]:
"""Return custom UI field behaviour for Redis connection."""
return {
"hidden_fields": ["schema", "extra"],
"relabeling": {},
}
@classmethod