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