From 05bc11779dda9edf419f49b6d28ed657975798d9 Mon Sep 17 00:00:00 2001 From: Antoni Ivanov Date: Thu, 14 Sep 2023 21:18:56 +0300 Subject: [PATCH] vdk-oracle: create oracle plugin Why? In order to support more use cases, vdk should support connecting and ingesting to an oracle database What? Add oracle plugin. Plugin supports simple queries, cli queries and ingestion. How was this tested? Local functional tests CI tests What kind of change is this? Feature/non-breaking Signed-off-by: Dilyan Marinov --- .../vdk-plugins/vdk-oracle/.plugin-ci.yml | 22 +++ projects/vdk-plugins/vdk-oracle/README.md | 111 +++++++++++ .../vdk-plugins/vdk-oracle/requirements.txt | 8 + projects/vdk-plugins/vdk-oracle/setup.py | 40 ++++ .../src/vdk/plugin/oracle/ingest_to_oracle.py | 165 ++++++++++++++++ .../vdk/plugin/oracle/oracle_configuration.py | 49 +++++ .../vdk/plugin/oracle/oracle_connection.py | 56 ++++++ .../src/vdk/plugin/oracle/oracle_plugin.py | 91 +++++++++ .../vdk-plugins/vdk-oracle/tests/conftest.py | 64 ++++++ .../00_drop_table.sql | 4 + .../10_create_table.sql | 1 + .../20_populate_table.sql | 1 + .../oracle-ingest-job-blob/00_drop_table.sql | 4 + .../10_create_table.sql | 4 + .../jobs/oracle-ingest-job-blob/20_ingest.py | 16 ++ .../00_drop_table.sql | 4 + .../10_ingest.py | 58 ++++++ .../00_drop_table.sql | 4 + .../10_create_table.sql | 8 + .../20_ingest.py | 58 ++++++ .../00_drop_table.sql | 4 + .../oracle-ingest-job-no-table/10_ingest.py | 48 +++++ .../jobs/oracle-ingest-job/00_drop_table.sql | 4 + .../oracle-ingest-job/10_create_table.sql | 9 + .../tests/jobs/oracle-ingest-job/20_ingest.py | 48 +++++ .../vdk-oracle/tests/test_plugin.py | 184 ++++++++++++++++++ 26 files changed, 1065 insertions(+) create mode 100644 projects/vdk-plugins/vdk-oracle/.plugin-ci.yml create mode 100644 projects/vdk-plugins/vdk-oracle/README.md create mode 100644 projects/vdk-plugins/vdk-oracle/requirements.txt create mode 100644 projects/vdk-plugins/vdk-oracle/setup.py create mode 100644 projects/vdk-plugins/vdk-oracle/src/vdk/plugin/oracle/ingest_to_oracle.py create mode 100644 projects/vdk-plugins/vdk-oracle/src/vdk/plugin/oracle/oracle_configuration.py create mode 100644 projects/vdk-plugins/vdk-oracle/src/vdk/plugin/oracle/oracle_connection.py create mode 100644 projects/vdk-plugins/vdk-oracle/src/vdk/plugin/oracle/oracle_plugin.py create mode 100644 projects/vdk-plugins/vdk-oracle/tests/conftest.py create mode 100644 projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-connect-execute-job/00_drop_table.sql create mode 100644 projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-connect-execute-job/10_create_table.sql create mode 100644 projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-connect-execute-job/20_populate_table.sql create mode 100644 projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-blob/00_drop_table.sql create mode 100644 projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-blob/10_create_table.sql create mode 100644 projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-blob/20_ingest.py create mode 100644 projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-different-payloads-no-table/00_drop_table.sql create mode 100644 projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-different-payloads-no-table/10_ingest.py create mode 100644 projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-different-payloads/00_drop_table.sql create mode 100644 projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-different-payloads/10_create_table.sql create mode 100644 projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-different-payloads/20_ingest.py create mode 100644 projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-no-table/00_drop_table.sql create mode 100644 projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-no-table/10_ingest.py create mode 100644 projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job/00_drop_table.sql create mode 100644 projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job/10_create_table.sql create mode 100644 projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job/20_ingest.py create mode 100644 projects/vdk-plugins/vdk-oracle/tests/test_plugin.py diff --git a/projects/vdk-plugins/vdk-oracle/.plugin-ci.yml b/projects/vdk-plugins/vdk-oracle/.plugin-ci.yml new file mode 100644 index 0000000000..6847e3f9a0 --- /dev/null +++ b/projects/vdk-plugins/vdk-oracle/.plugin-ci.yml @@ -0,0 +1,22 @@ +# Copyright 2021-2023 VMware, Inc. +# SPDX-License-Identifier: Apache-2.0 + +image: "python:3.7" + +.build-vdk-oracle: + variables: + PLUGIN_NAME: vdk-oracle + extends: .build-plugin-dind + +build-py37-vdk-oracle: + extends: .build-vdk-oracle + image: "python:3.7" + +build-py311-vdk-oracle: + extends: .build-vdk-oracle + image: "python:3.11" + +release-vdk-oracle: + variables: + PLUGIN_NAME: vdk-oracle + extends: .release-plugin diff --git a/projects/vdk-plugins/vdk-oracle/README.md b/projects/vdk-plugins/vdk-oracle/README.md new file mode 100644 index 0000000000..f7e7639c62 --- /dev/null +++ b/projects/vdk-plugins/vdk-oracle/README.md @@ -0,0 +1,111 @@ +# oracle + +Support for VDK Managed Oracle connection + +TODO: what the project is about, what is its purpose + + +## Usage + +``` +pip install vdk-oracle +``` + +### Configuration + +(`vdk config-help` is useful command to browse all config options of your installation of vdk) + +| Name | Description | (example) Value | +|--------------------------|--------------------------------------------------|----------------------| +| oracle_user | Username used when connecting to Oracle database | "my_user" | +| oracle_password | Password used when connecting to Oracle database | "super_secret_shhhh" | +| oracle_connection_string | The Oracle connection string | "localhost/free" | + +### Example + +#### Ingestion + +```python +import datetime +from decimal import Decimal + +def run(job_input): + + # Ingest object + payload_with_types = { + "id": 5, + "str_data": "string", + "int_data": 12, + "float_data": 1.2, + "bool_data": True, + "timestamp_data": datetime.datetime.fromtimestamp(1700554373), + "decimal_data": Decimal(0.1), + } + + job_input.send_object_for_ingestion( + payload=payload_with_types, destination_table="test_table" + ) + + # Ingest tabular data + col_names = [ + "id", + "str_data", + "int_data", + "float_data", + "bool_data", + "timestamp_data", + "decimal_data", + ] + row_data = [ + [ + 0, + "string", + 12, + 1.2, + True, + datetime.datetime.fromtimestamp(1700554373), + Decimal(1.1), + ], + [ + 1, + "string", + 12, + 1.2, + True, + datetime.datetime.fromtimestamp(1700554373), + Decimal(1.1), + ], + [ + 2, + "string", + 12, + 1.2, + True, + datetime.datetime.fromtimestamp(1700554373), + Decimal(1.1), + ], + ] + job_input.send_tabular_data_for_ingestion( + rows=row_data, column_names=col_names, destination_table="test_table" + ) +``` +### Build and testing + +``` +pip install -r requirements.txt +pip install -e . +pytest +``` + +In VDK repo [../build-plugin.sh](https://github.com/vmware/versatile-data-kit/tree/main/projects/vdk-plugins/build-plugin.sh) script can be used also. + + +#### Note about the CICD: + +.plugin-ci.yaml is needed only for plugins part of [Versatile Data Kit Plugin repo](https://github.com/vmware/versatile-data-kit/tree/main/projects/vdk-plugins). + +The CI/CD is separated in two stages, a build stage and a release stage. +The build stage is made up of a few jobs, all which inherit from the same +job configuration and only differ in the Python version they use (3.7, 3.8, 3.9 and 3.10). +They run according to rules, which are ordered in a way such that changes to a +plugin's directory trigger the plugin CI, but changes to a different plugin does not. diff --git a/projects/vdk-plugins/vdk-oracle/requirements.txt b/projects/vdk-plugins/vdk-oracle/requirements.txt new file mode 100644 index 0000000000..e4742c16c7 --- /dev/null +++ b/projects/vdk-plugins/vdk-oracle/requirements.txt @@ -0,0 +1,8 @@ +# this file is used to provide testing requirements +# for requirements (dependencies) needed during and after installation of the plugin see (and update) setup.py install_requires section + + +pytest +testcontainers +vdk-core +vdk-test-utils diff --git a/projects/vdk-plugins/vdk-oracle/setup.py b/projects/vdk-plugins/vdk-oracle/setup.py new file mode 100644 index 0000000000..1eac5dd10c --- /dev/null +++ b/projects/vdk-plugins/vdk-oracle/setup.py @@ -0,0 +1,40 @@ +# Copyright 2021-2023 VMware, Inc. +# SPDX-License-Identifier: Apache-2.0 +import pathlib + +import setuptools + +""" +Builds a package with the help of setuptools in order for this package to be imported in other projects +""" + +__version__ = "0.1.0" + +setuptools.setup( + name="vdk-oracle", + version=__version__, + url="https://github.com/vmware/versatile-data-kit", + description="Support for VDK Managed Oracle connection", + long_description=pathlib.Path("README.md").read_text(), + long_description_content_type="text/markdown", + install_requires=["vdk-core", "oracledb", "tabulate"], + package_dir={"": "src"}, + packages=setuptools.find_namespace_packages(where="src"), + # This is the only vdk plugin specifc part + # Define entry point called "vdk.plugin.run" with name of plugin and module to act as entry point. + entry_points={"vdk.plugin.run": ["vdk-oracle = vdk.plugin.oracle.oracle_plugin"]}, + classifiers=[ + "Development Status :: 2 - Pre-Alpha", + "License :: OSI Approved :: Apache Software License", + "Programming Language :: Python :: 3.7", + "Programming Language :: Python :: 3.8", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + ], + project_urls={ + "Documentation": "https://github.com/vmware/versatile-data-kit/tree/main/projects/vdk-plugins/vdk-oracle", + "Source Code": "https://github.com/vmware/versatile-data-kit/tree/main/projects/vdk-plugins/vdk-oracle", + "Bug Tracker": "https://github.com/vmware/versatile-data-kit/issues/new/choose", + }, +) diff --git a/projects/vdk-plugins/vdk-oracle/src/vdk/plugin/oracle/ingest_to_oracle.py b/projects/vdk-plugins/vdk-oracle/src/vdk/plugin/oracle/ingest_to_oracle.py new file mode 100644 index 0000000000..660ec5e912 --- /dev/null +++ b/projects/vdk-plugins/vdk-oracle/src/vdk/plugin/oracle/ingest_to_oracle.py @@ -0,0 +1,165 @@ +# Copyright 2021-2023 VMware, Inc. +# SPDX-License-Identifier: Apache-2.0 +import datetime +import logging +from decimal import Decimal +from typing import Any +from typing import Dict +from typing import List +from typing import Optional +from typing import Set + +from vdk.api.plugin.plugin_input import PEP249Connection +from vdk.internal.builtin_plugins.connection.impl.router import ManagedConnectionRouter +from vdk.internal.builtin_plugins.connection.managed_cursor import ManagedCursor +from vdk.internal.builtin_plugins.ingestion.ingester_base import IIngesterPlugin + +log = logging.getLogger(__name__) + + +class IngestToOracle(IIngesterPlugin): + def __init__(self, connections: ManagedConnectionRouter): + self.conn: PEP249Connection = connections.open_connection("ORACLE").connect() + self.cursor: ManagedCursor = self.conn.cursor() + self.table_cache: Set[str] = set() # Cache to store existing tables + self.column_cache: Dict[str, str] = {} # New cache for columns + + @staticmethod + def _get_oracle_type(value: Any) -> str: + type_mappings = { + int: "NUMBER", + float: "FLOAT", + Decimal: "DECIMAL(14, 8)", + str: "VARCHAR2(255)", + datetime.datetime: "TIMESTAMP", + bool: "NUMBER(1)", + bytes: "BLOB", + } + return type_mappings.get(type(value), "VARCHAR2(255)") + + def _table_exists(self, table_name: str) -> bool: + if table_name.upper() in self.table_cache: + return True + + self.cursor.execute( + f"SELECT COUNT(*) FROM user_tables WHERE table_name = :1", + [table_name.upper()], + ) + exists = bool(self.cursor.fetchone()[0]) + + if exists: + self.table_cache.add(table_name.upper()) + + return exists + + def _create_table(self, table_name: str, row: Dict[str, Any]) -> None: + column_defs = [f"{col} {self._get_oracle_type(row[col])}" for col in row.keys()] + create_table_sql = ( + f"CREATE TABLE {table_name.upper()} ({', '.join(column_defs)})" + ) + self.cursor.execute(create_table_sql) + + def _cache_columns(self, table_name: str) -> None: + try: + self.cursor.execute( + f"SELECT column_name FROM user_tab_columns WHERE table_name = '{table_name.upper()}'" + ) + result = self.cursor.fetchall() + self.column_cache[table_name.upper()] = {column[0] for column in result} + except Exception as e: + # TODO: https://github.com/vmware/versatile-data-kit/issues/2932 + log.error( + "An exception occurred while trying to cache columns. Ignoring for now." + ) + log.exception(e) + + def _add_columns(self, table_name: str, payload: List[Dict[str, Any]]) -> None: + if table_name.upper() not in self.column_cache: + self._cache_columns(table_name) + + existing_columns = self.column_cache[table_name.upper()] + + # Find unique new columns from all rows in the payload + all_columns = {col.upper() for row in payload for col in row.keys()} + new_columns = all_columns - existing_columns + + if new_columns: + column_defs = [] + for col in new_columns: + sample_value = next( + (row[col] for row in payload if row.get(col) is not None), None + ) + column_type = ( + self._get_oracle_type(sample_value) + if sample_value is not None + else "VARCHAR2(255)" + ) + column_defs.append(f"{col} {column_type}") + + alter_sql = ( + f"ALTER TABLE {table_name.upper()} ADD ({', '.join(column_defs)})" + ) + self.cursor.execute(alter_sql) + self.column_cache[table_name.upper()].update(new_columns) + + # TODO: https://github.com/vmware/versatile-data-kit/issues/2929 + # TODO: https://github.com/vmware/versatile-data-kit/issues/2930 + def _cast_to_correct_type(self, value: Any) -> Any: + if type(value) is Decimal: + return float(value) + return value + + # TODO: Look into potential optimizations + # TODO: https://github.com/vmware/versatile-data-kit/issues/2931 + def _insert_data(self, table_name: str, payload: List[Dict[str, Any]]) -> None: + if not payload: + return + + # group dicts by key set + batches = {} + for p in payload: + batch = frozenset(p.keys()) + if batch not in batches: + batches[batch] = [] + batches[batch].append(p) + + # create queries for groups of dicts with the same key set + queries = [] + batch_data = [] + for column_names, batch in batches.items(): + columns = list(column_names) + insert_sql = f"INSERT INTO {table_name} ({', '.join(columns)}) VALUES ({', '.join([':' + str(i + 1) for i in range(len(columns))])})" + queries.append(insert_sql) + temp_data = [] + for row in batch: + temp = [self._cast_to_correct_type(row[col]) for col in columns] + temp_data.append(temp) + batch_data.append(temp_data) + + # batch execute queries for dicts with the same key set + for i in range(len(queries)): + self.cursor.executemany(queries[i], batch_data[i]) + + def ingest_payload( + self, + payload: List[Dict[str, Any]], + destination_table: Optional[str] = None, + target: str = None, + collection_id: Optional[str] = None, + metadata: Optional[IIngesterPlugin.IngestionMetadata] = None, + ) -> None: + if not payload: + return None + if not destination_table: + raise ValueError("Destination table must be specified if not in payload.") + + if not self._table_exists(destination_table): + self._create_table(destination_table, payload[0]) + self._cache_columns(destination_table) + + self._add_columns(destination_table, payload) + self._insert_data(destination_table, payload) + + # TODO: test if we need this commit statement (most probably we don't, the connection already commits after every transaction) + self.conn.commit() + return metadata diff --git a/projects/vdk-plugins/vdk-oracle/src/vdk/plugin/oracle/oracle_configuration.py b/projects/vdk-plugins/vdk-oracle/src/vdk/plugin/oracle/oracle_configuration.py new file mode 100644 index 0000000000..497d904ada --- /dev/null +++ b/projects/vdk-plugins/vdk-oracle/src/vdk/plugin/oracle/oracle_configuration.py @@ -0,0 +1,49 @@ +# Copyright 2021-2023 VMware, Inc. +# SPDX-License-Identifier: Apache-2.0 +import json +import pathlib +import tempfile +from typing import Dict +from typing import Optional + +from vdk.internal.core.config import Configuration +from vdk.internal.core.config import ConfigurationBuilder + +ORACLE_USER = "ORACLE_USER" +ORACLE_PASSWORD = "ORACLE_PASSWORD" +ORACLE_CONNECTION_STRING = "ORACLE_CONNECTION_STRING" + + +class OracleConfiguration: + def __init__(self, configuration: Configuration): + self.__config = configuration + + def get_oracle_user(self) -> str: + return self.__config.get_value(ORACLE_USER) + + def get_oracle_password(self) -> str: + return self.__config.get_value(ORACLE_PASSWORD) + + def get_oracle_connection_string(self) -> Optional[Dict[str, str]]: + return self.__config.get_value(ORACLE_CONNECTION_STRING) + + @staticmethod + def add_definitions(config_builder: ConfigurationBuilder): + config_builder.add( + key=ORACLE_USER, + default_value=None, + is_sensitive=True, + description="The Oracle user for the database connection", + ) + config_builder.add( + key=ORACLE_PASSWORD, + default_value=None, + is_sensitive=True, + description="The oracle password for the database connection", + ) + config_builder.add( + key=ORACLE_CONNECTION_STRING, + default_value=None, + is_sensitive=True, + description="The Oracle database connection string", + ) diff --git a/projects/vdk-plugins/vdk-oracle/src/vdk/plugin/oracle/oracle_connection.py b/projects/vdk-plugins/vdk-oracle/src/vdk/plugin/oracle/oracle_connection.py new file mode 100644 index 0000000000..6b5cdc5c95 --- /dev/null +++ b/projects/vdk-plugins/vdk-oracle/src/vdk/plugin/oracle/oracle_connection.py @@ -0,0 +1,56 @@ +# Copyright 2021-2023 VMware, Inc. +# SPDX-License-Identifier: Apache-2.0 +import logging +from typing import Any +from typing import List + +from vdk.internal.builtin_plugins.connection.managed_connection_base import ( + ManagedConnectionBase, +) +from vdk.internal.builtin_plugins.connection.pep249.interfaces import PEP249Connection + +log = logging.getLogger(__name__) + + +class OracleConnection(ManagedConnectionBase): + def __init__(self, user: str, password: str, connection_string: str): + super().__init__(log) + self._oracle_user = user + self._oracle_password = password + self._oracle_connection_string = connection_string + + def _connect(self) -> PEP249Connection: + import oracledb + + conn = oracledb.connect( + user=self._oracle_user, + password=self._oracle_password, + dsn=self._oracle_connection_string, + ) + return conn + + def _is_connected(self) -> bool: + if None is self._is_db_con_open: + return False + if False is self._is_db_con_open: + return False + try: + """ + The remote end (the database server) may have disconnected or the session may have timed out (on the remote one) + but in the client (here in vdk - we are the client) we do not know. + We try to check with "select count(*) from user_tables where rownum = 1". + Note: The default behavior is to just use "select 1", but this didn't work for Oracle + """ + self._cursor().execute("select count(*) from user_tables where rownum = 1") + return True + except Exception as e: + self._log.debug( + f"Connection {self} is disconnected ('select 1' returned {e})" + ) + return False + + def execute_query(self, query: str) -> List[List[Any]]: + try: + return super().execute_query(query) + finally: + self.commit() diff --git a/projects/vdk-plugins/vdk-oracle/src/vdk/plugin/oracle/oracle_plugin.py b/projects/vdk-plugins/vdk-oracle/src/vdk/plugin/oracle/oracle_plugin.py new file mode 100644 index 0000000000..68f0dcc6ab --- /dev/null +++ b/projects/vdk-plugins/vdk-oracle/src/vdk/plugin/oracle/oracle_plugin.py @@ -0,0 +1,91 @@ +# Copyright 2021-2023 VMware, Inc. +# SPDX-License-Identifier: Apache-2.0 +import logging +from typing import List + +import click +import oracledb +from tabulate import tabulate +from vdk.api.plugin.hook_markers import hookimpl +from vdk.api.plugin.plugin_registry import IPluginRegistry +from vdk.internal.builtin_plugins.run.job_context import JobContext +from vdk.internal.core.config import ConfigurationBuilder +from vdk.internal.util.decorators import closing_noexcept_on_close +from vdk.plugin.oracle.ingest_to_oracle import IngestToOracle +from vdk.plugin.oracle.oracle_configuration import ORACLE_CONNECTION_STRING +from vdk.plugin.oracle.oracle_configuration import ORACLE_PASSWORD +from vdk.plugin.oracle.oracle_configuration import ORACLE_USER +from vdk.plugin.oracle.oracle_configuration import OracleConfiguration +from vdk.plugin.oracle.oracle_connection import OracleConnection + +""" +Include the plugins implementation. For example: +""" + +log = logging.getLogger(__name__) + + +class OraclePlugin: + @hookimpl(tryfirst=True) + def vdk_configure(self, config_builder: ConfigurationBuilder): + OracleConfiguration.add_definitions(config_builder) + + @hookimpl + def initialize_job(self, context: JobContext): + conf = OracleConfiguration(context.core_context.configuration) + context.connections.add_open_connection_factory_method( + "ORACLE", + lambda: OracleConnection( + conf.get_oracle_user(), + conf.get_oracle_password(), + conf.get_oracle_connection_string(), + ), + ) + context.ingester.add_ingester_factory_method( + "oracle", (lambda: IngestToOracle(context.connections)) + ) + + +@hookimpl +def vdk_start(plugin_registry: IPluginRegistry, command_line_args: List): + plugin_registry.load_plugin_with_hooks_impl(OraclePlugin(), "OraclePlugin") + + +# TODO: https://github.com/vmware/versatile-data-kit/issues/2940 +@click.command( + name="oracle-query", + help="Execute an Oracle query against an Oracle database (should be configured with env variables)", +) +@click.option("-q", "--query", type=click.STRING, required=True) +@click.pass_context +def oracle_query(ctx: click.Context, query): + conf = ctx.obj.configuration + conn = oracledb.connect( + user=conf.get_value(ORACLE_USER), + password=conf.get_value(ORACLE_PASSWORD), + dsn=conf.get_value(ORACLE_CONNECTION_STRING), + ) + + with closing_noexcept_on_close(conn.cursor()) as cursor: + cursor.execute(query) + column_names = ( + [column_info[0] for column_info in cursor.description] + if cursor.description + else () # same as the default value for the headers parameters of the tabulate function + ) + try: + res = cursor.fetchall() + click.echo(tabulate(res, headers=column_names)) + except Exception as e: + if str(e) == "DPY-1003: the executed statement does not return rows": + log.info( + "Query did not produce results, e.g. DROP TABLE, but was successful" + ) + else: + raise e + conn.commit() + + +@hookimpl +def vdk_command_line(root_command: click.Group): + root_command.add_command(oracle_query) diff --git a/projects/vdk-plugins/vdk-oracle/tests/conftest.py b/projects/vdk-plugins/vdk-oracle/tests/conftest.py new file mode 100644 index 0000000000..3c0617cdf9 --- /dev/null +++ b/projects/vdk-plugins/vdk-oracle/tests/conftest.py @@ -0,0 +1,64 @@ +# Copyright 2021-2023 VMware, Inc. +# SPDX-License-Identifier: Apache-2.0 +import os +import time +from unittest import mock + +import pytest +from testcontainers.core.container import DockerContainer +from testcontainers.core.waiting_utils import wait_for_logs + +ORACLE_IMAGE = "container-registry.oracle.com/database/free:latest" + +DB_DEFAULT_TYPE = "DB_DEFAULT_TYPE" +ORACLE_USER = "ORACLE_USER" +ORACLE_PASSWORD = "ORACLE_PASSWORD" +ORACLE_CONNECTION_STRING = "ORACLE_CONNECTION_STRING" +VDK_LOG_EXECUTION_RESULT = "VDK_LOG_EXECUTION_RESULT" +VDK_INGEST_METHOD_DEFAULT = "VDK_INGEST_METHOD_DEFAULT" + + +@pytest.fixture(scope="session") +@mock.patch.dict( + os.environ, + { + DB_DEFAULT_TYPE: "oracle", + ORACLE_USER: "SYSTEM", + ORACLE_PASSWORD: "Gr0mh3llscr3am", + ORACLE_CONNECTION_STRING: "localhost:1521/FREE", + VDK_LOG_EXECUTION_RESULT: "True", + VDK_INGEST_METHOD_DEFAULT: "ORACLE", + }, +) +def oracle_db(request): + port = 1521 + password = os.environ[ORACLE_PASSWORD] + container = ( + DockerContainer(ORACLE_IMAGE) + .with_bind_ports(port, port) + .with_env("ORACLE_PWD", password) + .with_env("ORACLE_CHARACTERSET", "UTF8") + ) + try: + container.start() + wait_for_logs( + container, + "DATABASE IS READY TO USE", + timeout=120, + ) + time.sleep(2) + print( + f"Oracle db started on port {container.get_exposed_port(port)} and host {container.get_container_host_ip()}" + ) + except Exception as e: + print(f"Failed to start Oracle DB: {e}") + print(f"Container logs: {container.get_logs()}") + raise e + + def stop_container(): + container.stop() + print("Oracle DB stopped") + + request.addfinalizer(stop_container) + + return container diff --git a/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-connect-execute-job/00_drop_table.sql b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-connect-execute-job/00_drop_table.sql new file mode 100644 index 0000000000..dd40f2d142 --- /dev/null +++ b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-connect-execute-job/00_drop_table.sql @@ -0,0 +1,4 @@ +begin + execute immediate 'drop table todoitem'; + exception when others then if sqlcode <> -942 then raise; end if; +end; diff --git a/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-connect-execute-job/10_create_table.sql b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-connect-execute-job/10_create_table.sql new file mode 100644 index 0000000000..dca6b6f969 --- /dev/null +++ b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-connect-execute-job/10_create_table.sql @@ -0,0 +1 @@ +create table todoitem (id number generated always as identity,description varchar2(4000),done number(1,0),primary key (id)) diff --git a/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-connect-execute-job/20_populate_table.sql b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-connect-execute-job/20_populate_table.sql new file mode 100644 index 0000000000..9f45cd536e --- /dev/null +++ b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-connect-execute-job/20_populate_table.sql @@ -0,0 +1 @@ +insert into todoitem (description, done) values('Task 1', 1) diff --git a/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-blob/00_drop_table.sql b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-blob/00_drop_table.sql new file mode 100644 index 0000000000..865c2f0050 --- /dev/null +++ b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-blob/00_drop_table.sql @@ -0,0 +1,4 @@ +begin + execute immediate 'drop table test_table'; + exception when others then if sqlcode <> -942 then raise; end if; +end; diff --git a/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-blob/10_create_table.sql b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-blob/10_create_table.sql new file mode 100644 index 0000000000..211bf1e8a9 --- /dev/null +++ b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-blob/10_create_table.sql @@ -0,0 +1,4 @@ +create table test_table ( + id number, + blob_data BLOB, + primary key(id)) diff --git a/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-blob/20_ingest.py b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-blob/20_ingest.py new file mode 100644 index 0000000000..cd6c499c8c --- /dev/null +++ b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-blob/20_ingest.py @@ -0,0 +1,16 @@ +# Copyright 2021-2023 VMware, Inc. +# SPDX-License-Identifier: Apache-2.0 +def run(job_input): + frost = """The woods are lovely, dark and deep, +But I have promises to keep, +And miles to go before I sleep, +And miles to go before I sleep. +""" + payload_with_blob = { + "id": 5, + "blob_data": frost.encode("utf-8"), + } + + job_input.send_object_for_ingestion( + payload=payload_with_blob, destination_table="test_table" + ) diff --git a/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-different-payloads-no-table/00_drop_table.sql b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-different-payloads-no-table/00_drop_table.sql new file mode 100644 index 0000000000..865c2f0050 --- /dev/null +++ b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-different-payloads-no-table/00_drop_table.sql @@ -0,0 +1,4 @@ +begin + execute immediate 'drop table test_table'; + exception when others then if sqlcode <> -942 then raise; end if; +end; diff --git a/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-different-payloads-no-table/10_ingest.py b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-different-payloads-no-table/10_ingest.py new file mode 100644 index 0000000000..2c853e97df --- /dev/null +++ b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-different-payloads-no-table/10_ingest.py @@ -0,0 +1,58 @@ +# Copyright 2021-2023 VMware, Inc. +# SPDX-License-Identifier: Apache-2.0 +import datetime + + +def run(job_input): + payloads = [ + { + "id": 0, + }, + { + "id": 1, + "str_data": "string", + }, + { + "id": 2, + "str_data": "string", + "int_data": 12, + }, + { + "id": 3, + "str_data": "string", + "int_data": 12, + "float_data": 1.2, + }, + { + "id": 4, + "str_data": "string", + "int_data": 12, + "float_data": 1.2, + "bool_data": True, + }, + { + "id": 5, + "str_data": "string", + "int_data": 12, + "float_data": 1.2, + "bool_data": True, + "timestamp_data": datetime.datetime.fromtimestamp(1700554373), + }, + { + "id": 6, + "str_data": "string", + "int_data": 12, + "float_data": 1.2, + }, + { + "id": 7, + "str_data": "string", + "int_data": 12, + "float_data": 1.2, + "bool_data": True, + }, + ] + for payload in payloads: + job_input.send_object_for_ingestion( + payload=payload, destination_table="test_table" + ) diff --git a/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-different-payloads/00_drop_table.sql b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-different-payloads/00_drop_table.sql new file mode 100644 index 0000000000..865c2f0050 --- /dev/null +++ b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-different-payloads/00_drop_table.sql @@ -0,0 +1,4 @@ +begin + execute immediate 'drop table test_table'; + exception when others then if sqlcode <> -942 then raise; end if; +end; diff --git a/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-different-payloads/10_create_table.sql b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-different-payloads/10_create_table.sql new file mode 100644 index 0000000000..eac5bed1e7 --- /dev/null +++ b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-different-payloads/10_create_table.sql @@ -0,0 +1,8 @@ +create table test_table ( + id number, + str_data varchar2(255), + int_data number, + float_data float, + bool_data number(1), + timestamp_data timestamp, + primary key(id)) diff --git a/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-different-payloads/20_ingest.py b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-different-payloads/20_ingest.py new file mode 100644 index 0000000000..2c853e97df --- /dev/null +++ b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-different-payloads/20_ingest.py @@ -0,0 +1,58 @@ +# Copyright 2021-2023 VMware, Inc. +# SPDX-License-Identifier: Apache-2.0 +import datetime + + +def run(job_input): + payloads = [ + { + "id": 0, + }, + { + "id": 1, + "str_data": "string", + }, + { + "id": 2, + "str_data": "string", + "int_data": 12, + }, + { + "id": 3, + "str_data": "string", + "int_data": 12, + "float_data": 1.2, + }, + { + "id": 4, + "str_data": "string", + "int_data": 12, + "float_data": 1.2, + "bool_data": True, + }, + { + "id": 5, + "str_data": "string", + "int_data": 12, + "float_data": 1.2, + "bool_data": True, + "timestamp_data": datetime.datetime.fromtimestamp(1700554373), + }, + { + "id": 6, + "str_data": "string", + "int_data": 12, + "float_data": 1.2, + }, + { + "id": 7, + "str_data": "string", + "int_data": 12, + "float_data": 1.2, + "bool_data": True, + }, + ] + for payload in payloads: + job_input.send_object_for_ingestion( + payload=payload, destination_table="test_table" + ) diff --git a/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-no-table/00_drop_table.sql b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-no-table/00_drop_table.sql new file mode 100644 index 0000000000..865c2f0050 --- /dev/null +++ b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-no-table/00_drop_table.sql @@ -0,0 +1,4 @@ +begin + execute immediate 'drop table test_table'; + exception when others then if sqlcode <> -942 then raise; end if; +end; diff --git a/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-no-table/10_ingest.py b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-no-table/10_ingest.py new file mode 100644 index 0000000000..3e80004310 --- /dev/null +++ b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job-no-table/10_ingest.py @@ -0,0 +1,48 @@ +# Copyright 2021-2023 VMware, Inc. +# SPDX-License-Identifier: Apache-2.0 +import datetime +from decimal import Decimal + + +def run(job_input): + col_names = [ + "id", + "str_data", + "int_data", + "float_data", + "bool_data", + "timestamp_data", + "decimal_data", + ] + row_data = [ + [ + 0, + "string", + 12, + 1.2, + True, + datetime.datetime.fromtimestamp(1700554373), + Decimal(1.1), + ], + [ + 1, + "string", + 12, + 1.2, + True, + datetime.datetime.fromtimestamp(1700554373), + Decimal(1.1), + ], + [ + 2, + "string", + 12, + 1.2, + True, + datetime.datetime.fromtimestamp(1700554373), + Decimal(1.1), + ], + ] + job_input.send_tabular_data_for_ingestion( + rows=row_data, column_names=col_names, destination_table="test_table" + ) diff --git a/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job/00_drop_table.sql b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job/00_drop_table.sql new file mode 100644 index 0000000000..865c2f0050 --- /dev/null +++ b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job/00_drop_table.sql @@ -0,0 +1,4 @@ +begin + execute immediate 'drop table test_table'; + exception when others then if sqlcode <> -942 then raise; end if; +end; diff --git a/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job/10_create_table.sql b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job/10_create_table.sql new file mode 100644 index 0000000000..0d8694fe06 --- /dev/null +++ b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job/10_create_table.sql @@ -0,0 +1,9 @@ +create table test_table ( + id number, + str_data varchar2(255), + int_data number, + float_data float, + bool_data number(1), + timestamp_data timestamp, + decimal_data decimal(14,8), + primary key(id)) diff --git a/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job/20_ingest.py b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job/20_ingest.py new file mode 100644 index 0000000000..dd85621272 --- /dev/null +++ b/projects/vdk-plugins/vdk-oracle/tests/jobs/oracle-ingest-job/20_ingest.py @@ -0,0 +1,48 @@ +# Copyright 2021-2023 VMware, Inc. +# SPDX-License-Identifier: Apache-2.0 +import datetime +from decimal import Decimal + + +def run(job_input): + # TODO: https://github.com/vmware/versatile-data-kit/issues/2929 + # setup different data types (all passed initially as strings) are cast correctly + # payload = { + # "id": "", + # "str_data": "string", + # "int_data": "12", + # "float_data": "1.2", + # "bool_data": "True", + # # TODO: add timestamp + # # TODO: add decimal + # } + + # for i in range(5): + # local_payload = payload.copy() + # local_payload["id"] = i + # job_input.send_object_for_ingestion( + # payload=local_payload, destination_table="test_table" + # ) + + payload_with_types = { + "id": 5, + "str_data": "string", + "int_data": 12, + "float_data": 1.2, + "bool_data": True, + "timestamp_data": datetime.datetime.fromtimestamp(1700554373), + "decimal_data": Decimal(0.1), + } + + job_input.send_object_for_ingestion( + payload=payload_with_types, destination_table="test_table" + ) + + # TODO: https://github.com/vmware/versatile-data-kit/issues/2930 + # this setup: + # a) partial payload (only few columns are included) + # b) includes float data which is NaN + # payload2 = {"id": 6, "float_data": math.nan, "int_data": math.nan} + # job_input.send_object_for_ingestion( + # payload=payload2, destination_table="test_table" + # ) diff --git a/projects/vdk-plugins/vdk-oracle/tests/test_plugin.py b/projects/vdk-plugins/vdk-oracle/tests/test_plugin.py new file mode 100644 index 0000000000..7810f8f9b6 --- /dev/null +++ b/projects/vdk-plugins/vdk-oracle/tests/test_plugin.py @@ -0,0 +1,184 @@ +# Copyright 2021-2023 VMware, Inc. +# SPDX-License-Identifier: Apache-2.0 +import os +from unittest import mock +from unittest import TestCase + +import pytest +from click.testing import Result +from vdk.plugin.oracle import oracle_plugin +from vdk.plugin.test_utils.util_funcs import cli_assert_equal +from vdk.plugin.test_utils.util_funcs import CliEntryBasedTestRunner +from vdk.plugin.test_utils.util_funcs import jobs_path_from_caller_directory + +DB_DEFAULT_TYPE = "DB_DEFAULT_TYPE" +ORACLE_USER = "ORACLE_USER" +ORACLE_PASSWORD = "ORACLE_PASSWORD" +ORACLE_CONNECTION_STRING = "ORACLE_CONNECTION_STRING" +VDK_LOG_EXECUTION_RESULT = "VDK_LOG_EXECUTION_RESULT" +VDK_INGEST_METHOD_DEFAULT = "VDK_INGEST_METHOD_DEFAULT" + + +@pytest.mark.usefixtures("oracle_db") +@mock.patch.dict( + os.environ, + { + DB_DEFAULT_TYPE: "oracle", + ORACLE_USER: "SYSTEM", + ORACLE_PASSWORD: "Gr0mh3llscr3am", + ORACLE_CONNECTION_STRING: "localhost:1521/FREE", + VDK_LOG_EXECUTION_RESULT: "True", + VDK_INGEST_METHOD_DEFAULT: "ORACLE", + }, +) +class OracleTests(TestCase): + def test_oracle_connect_execute(self): + runner = CliEntryBasedTestRunner(oracle_plugin) + result: Result = runner.invoke( + ["run", jobs_path_from_caller_directory("oracle-connect-execute-job")] + ) + cli_assert_equal(0, result) + _verify_query_execution(runner) + + def test_oracle_ingest_existing_table(self): + runner = CliEntryBasedTestRunner(oracle_plugin) + result: Result = runner.invoke( + ["run", jobs_path_from_caller_directory("oracle-ingest-job")] + ) + cli_assert_equal(0, result) + _verify_ingest_execution(runner) + + def test_oracle_ingest_no_table(self): + runner = CliEntryBasedTestRunner(oracle_plugin) + result: Result = runner.invoke( + ["run", jobs_path_from_caller_directory("oracle-ingest-job-no-table")] + ) + cli_assert_equal(0, result) + _verify_ingest_execution_no_table(runner) + + def test_oracle_ingest_different_payloads(self): + runner = CliEntryBasedTestRunner(oracle_plugin) + result: Result = runner.invoke( + [ + "run", + jobs_path_from_caller_directory("oracle-ingest-job-different-payloads"), + ] + ) + cli_assert_equal(0, result) + _verify_ingest_execution_different_payloads(runner) + + def test_oracle_ingest_different_payloads_no_table(self): + runner = CliEntryBasedTestRunner(oracle_plugin) + result: Result = runner.invoke( + [ + "run", + jobs_path_from_caller_directory( + "oracle-ingest-job-different-payloads-no-table" + ), + ] + ) + cli_assert_equal(0, result) + _verify_ingest_execution_different_payloads_no_table(runner) + + def test_oracle_ingest_blob(self): + runner = CliEntryBasedTestRunner(oracle_plugin) + result: Result = runner.invoke( + [ + "run", + jobs_path_from_caller_directory("oracle-ingest-job-blob"), + ] + ) + cli_assert_equal(0, result) + _verify_ingest_blob(runner) + + +def _verify_query_execution(runner): + check_result = runner.invoke(["oracle-query", "--query", "SELECT * FROM todoitem"]) + expected = ( + " ID DESCRIPTION DONE\n" + "---- ------------- ------\n" + " 1 Task 1 1\n" + ) + assert check_result.output == expected + + +def _verify_ingest_execution(runner): + check_result = runner.invoke( + ["oracle-query", "--query", "SELECT * FROM test_table"] + ) + expected = ( + " ID STR_DATA INT_DATA FLOAT_DATA BOOL_DATA " + "TIMESTAMP_DATA DECIMAL_DATA\n" + "---- ---------- ---------- ------------ ----------- " + "------------------- --------------\n" + " 5 string 12 1.2 1 2023-11-21 " + "08:12:53 0.1\n" + ) + assert check_result.output == expected + + +def _verify_ingest_execution_no_table(runner): + check_result = runner.invoke( + ["oracle-query", "--query", "SELECT * FROM test_table"] + ) + expected = ( + " ID STR_DATA INT_DATA FLOAT_DATA BOOL_DATA " + "TIMESTAMP_DATA DECIMAL_DATA\n" + "---- ---------- ---------- ------------ ----------- " + "------------------- --------------\n" + " 0 string 12 1.2 1 " + "2023-11-21T08:12:53 1.1\n" + " 1 string 12 1.2 1 " + "2023-11-21T08:12:53 1.1\n" + " 2 string 12 1.2 1 " + "2023-11-21T08:12:53 1.1\n" + ) + assert check_result.output == expected + + +def _verify_ingest_execution_different_payloads_no_table(runner): + check_result = runner.invoke( + ["oracle-query", "--query", "SELECT count(*) FROM test_table"] + ) + expected = " COUNT(*)\n----------\n 8\n" + assert check_result.output == expected + + +def _verify_ingest_execution_different_payloads(runner): + check_result = runner.invoke( + ["oracle-query", "--query", "SELECT * FROM test_table"] + ) + expected = ( + " ID STR_DATA INT_DATA FLOAT_DATA BOOL_DATA TIMESTAMP_DATA\n" + "---- ---------- ---------- ------------ ----------- " + "-------------------\n" + " 0\n" + " 1 string\n" + " 2 string 12\n" + " 3 string 12 1.2\n" + " 6 string 12 1.2\n" + " 4 string 12 1.2 1\n" + " 7 string 12 1.2 1\n" + " 5 string 12 1.2 1 2023-11-21 " + "08:12:53\n" + ) + assert check_result.output == expected + + +def _verify_ingest_blob(runner): + check_result = runner.invoke( + [ + "oracle-query", + "--query", + "SELECT utl_raw.cast_to_varchar2(dbms_lob.substr(blob_data,2000,1)) FROM test_table", + ] + ) + expected = ( + "UTL_RAW.CAST_TO_VARCHAR2(DBMS_LOB.SUBSTR(BLOB_DATA,2000,1))\n" + "-------------------------------------------------------------\n" + "The woods are lovely, dark and deep,\n" + "But I have promises to keep,\n" + "And miles to go before I sleep,\n" + "And miles to go before I sleep.\n" + ) + assert check_result.output == expected