7-Objects.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![Banner](images/banner.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Oracle Database Objects and Collections\n",
    "\n",
    "Documentation reference link: [Fetching Oracle Database Objects and Collections](https://python-oracledb.readthedocs.io/en/latest/user_guide/sql_execution.html#fetching-oracle-database-objects-and-collections)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import oracledb"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "un = \"pythondemo\"\n",
    "pw = \"welcome\"\n",
    "cs = \"localhost/orclpdb1\"\n",
    "\n",
    "connection = oracledb.connect(user=un, password=pw, dsn=cs)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Binding Named Objects\n",
    "\n",
    "Create a demonstration table. This table uses the predefined SDO_GEOMETRY object which stores spatial information:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "with connection.cursor() as cursor:\n",
    "    try:\n",
    "        cursor.execute(\"drop table TestGeometry\")\n",
    "    except:\n",
    "        ;\n",
    "        \n",
    "    cursor.execute(\"\"\"create table TestGeometry (\n",
    "                      IntCol   number(9) not null,\n",
    "                      Geometry sdo_geometry not null)\"\"\")\n",
    "    \n",
    "print(\"Done\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Using python-oracledb functions like `gettype()` and `extend()` you can create a Python representation of the database object:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "with connection.cursor() as cursor:\n",
    "    \n",
    "    typeObj = connection.gettype(\"SDO_GEOMETRY\")\n",
    "    elementInfoTypeObj = connection.gettype(\"SDO_ELEM_INFO_ARRAY\")\n",
    "    ordinateTypeObj = connection.gettype(\"SDO_ORDINATE_ARRAY\")\n",
    "\n",
    "    obj = typeObj()               # Alternatively use 'obj = typeObj.newobject()''\n",
    "    obj.SDO_GTYPE = 2003\n",
    "    obj.SDO_ELEM_INFO = elementInfoTypeObj()\n",
    "    obj.SDO_ELEM_INFO.extend([1, 1003, 3])\n",
    "    obj.SDO_ORDINATES = ordinateTypeObj()\n",
    "    obj.SDO_ORDINATES.extend([1, 1, 5, 7])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Calling `gettype()` requires multiple round-trips to the database, so avoid calling it unnecessarily."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The new object can be bound directly for insertion:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "with connection.cursor() as cursor:\n",
    "    cursor.execute(\"insert into TestGeometry values (1, :objbv)\",  {\"objbv\": obj})\n",
    "    \n",
    "print(\"Done\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "And then fetched back:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "with connection.cursor() as cursor:\n",
    "    for (id, obj) in cursor.execute(\"select IntCol, Geometry from testgeometry\"):\n",
    "        print(id, obj)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Simple attribute access is easy:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "with connection.cursor() as cursor:\n",
    "    for (id, obj) in cursor.execute(\"select IntCol, Geometry from testgeometry\"):\n",
    "        print(\"SDO_GTYPE is\", obj.SDO_GTYPE)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    " To display all attributes, create a helper function:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Oracle Database object dumper\n",
    "\n",
    "def dumpobject(obj, prefix = \"  \"):\n",
    "    if obj.type.iscollection:\n",
    "        print(prefix, \"[\")\n",
    "        for value in obj.aslist():\n",
    "            if isinstance(value, oracledb.Object):\n",
    "                dumpobject(value, prefix + \"  \")\n",
    "            else:\n",
    "                print(prefix + \"  \", repr(value))\n",
    "        print(prefix, \"]\")\n",
    "    else:\n",
    "        print(prefix, \"{\")\n",
    "        for attr in obj.type.attributes:\n",
    "            value = getattr(obj, attr.name)\n",
    "            if isinstance(value, oracledb.Object):\n",
    "                print(prefix + \"  \" + attr.name + \" :\")\n",
    "                dumpobject(value, prefix + \"    \")\n",
    "            else:\n",
    "                print(prefix + \"  \" + attr.name + \" :\", repr(value))\n",
    "        print(prefix, \"}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Using the helper function shows the full object structure:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "with connection.cursor() as cursor:\n",
    "    for (id, obj) in cursor.execute(\"select IntCol, Geometry from testgeometry\"):\n",
    "        print(\"Id: \", id)\n",
    "        dumpobject(obj)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# PL/SQL Collections\n",
    "\n",
    "The sample schema uses PL/SQL collections"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "cursor = connection.cursor()\n",
    "\n",
    "cursor.execute(\"select dbms_metadata.get_ddl('PACKAGE', 'PKG_DEMO') from dual\")\n",
    "ddl, = cursor.fetchone()\n",
    "print(ddl.read())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To get a collection, create a Python variable with the database object type:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "typeObj = connection.gettype(\"PKG_DEMO.UDT_STRINGLIST\")\n",
    "obj = typeObj()\n",
    "\n",
    "# call the stored procedure which will populate the object\n",
    "cursor = connection.cursor()\n",
    "cursor.callproc(\"pkg_Demo.DemoCollectionOut\", (obj,))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To show the collection indexes and values:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ix = obj.first()\n",
    "while ix is not None:\n",
    "    print(ix, \"->\", obj.getelement(ix))\n",
    "    ix = obj.next(ix)\n",
    "print()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Show the values as a simple list:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(obj.aslist())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Show the values as a simple dictionary:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(obj.asdict())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Binding PL/SQL Records\n",
    "\n",
    "Create a new Python object of the correct type and set attribute values:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import datetime\n",
    "\n",
    "typeObj = connection.gettype(\"PKG_DEMO.UDT_DEMORECORD\")\n",
    "obj = typeObj()\n",
    "\n",
    "obj.NUMBERVALUE = 6\n",
    "obj.STRINGVALUE = \"Test String\"\n",
    "obj.DATEVALUE = datetime.datetime(2016, 5, 28)\n",
    "obj.BOOLEANVALUE = False"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Call the stored procedure which will modify the object:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "with connection.cursor() as cursor:\n",
    "    cursor.callproc(\"pkg_Demo.DemoRecordsInOut\", (obj,))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Show the modified values:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(\"NUMBERVALUE ->\", obj.NUMBERVALUE)\n",
    "print(\"STRINGVALUE ->\", obj.STRINGVALUE)\n",
    "print(\"DATEVALUE ->\", obj.DATEVALUE)\n",
    "print(\"BOOLEANVALUE ->\", obj.BOOLEANVALUE)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}