Using Oracle Database from Python and Zato Services

-- Table definition
CREATE TABLE users (
    user_id NUMBER,
    username VARCHAR2(50)
);

-- Sample data
INSERT INTO users (user_id, username) VALUES (1, 'john_doe');
INSERT INTO users (user_id, username) VALUES (2, 'jane_smith');
INSERT INTO users (user_id, username) VALUES (3, 'bob_jones');

-- Stored procedure: process_data
CREATE OR REPLACE PROCEDURE process_data (
    input_num   IN  NUMBER,
    input_str   IN  VARCHAR2,
    output_num  OUT NUMBER,
    output_str  OUT VARCHAR2
)
AS
BEGIN
    output_num := input_num * 2;
    output_str := 'Input was: ' || input_str;
END process_data;
/

-- Stored procedure: get_users
CREATE OR REPLACE PROCEDURE get_users (
    recordset OUT SYS_REFCURSOR
)
AS
BEGIN
    OPEN recordset FOR
        SELECT user_id, username
        FROM users
        ORDER BY user_id;
END get_users;
/

# -*- coding: utf-8 -*-
from zato.server.service import Service

class GetAllUsers(Service):
    """ Service to retrieve all users from the database.
    """
    def handle(self):

        # Obtain a reference to the configured Oracle Database connection
        conn = self.out.sql['My Oracle DB']

        # Define the SQL query to select all rows from the users table
        query = 'select * from users'

        # Execute the query; returns a list of dictionaries, one per row
        response = conn.execute(query)

        # Set the service response to the query result
        self.response.payload = response

[
  {"user_id":1, "username":"john.doe"},
  {"user_id":2, "username":"jane.smith"},
  {"user_id":3,"username":"bob.jones"}
]

# -*- coding: utf-8 -*-
from zato.server.service import Service

class GetUserById(Service):
    """ Service to fetch a user by their user_id.
    """

    def handle(self):

        # Get the Oracle Database connection from the pool
        conn = self.out.sql['My Oracle DB']

        # Parameterized SQL to prevent injection
        query = 'select * from users where user_id = :user_id'

        # In a real service, this would be read from incoming JSON
        params = {'user_id': 1}

        # Execute the query with parameters; returns a list
        response = conn.execute(query, params)

        # Set the result as the service's response
        self.response.payload = response

[{"user_id":1,"username":"john.doe"}]

# -*- coding: utf-8 -*-

# Zato
from zato.common.oracledb import NumberIn, NumberOut, StringIn, StringOut
from zato.server.service import Service

class CallProcessData(Service):
    """ Service to call a stored procedure with input/output params.
    """
    def handle(self):

        # Obtain Oracle Database connection
        conn = self.out.sql['My Oracle DB']

        # Prepare input parameter for NUMBER
        in_num = NumberIn(333)

        # Prepare input parameter for VARCHAR2
        in_str = StringIn('Hello')

        # Prepare output parameter for NUMBER (will be written to by the procedure)
        out_num = NumberOut()

        # Prepare output parameter for VARCHAR2, specify max buffer size (optionally)
        out_str = StringOut(size=200)

        # Build the parameter list in the order expected by the procedure
        params = [in_num, in_str, out_num, out_str]

        # Call the stored procedure with the parameters
        response = conn.callproc('process_data', params)

        # Return the output values as a dictionary in the response
        self.response.payload = {
            'output_num': out_num.get(),
            'output_str': out_str.get()
        }

{"output_num":666, "output_str":"Input was: Hello"}

# -*- coding: utf-8 -*-
from zato.common.oracledb import RowsOut
from zato.server.service import Service

class CallGetUsers(Service):
    """ Service to call a procedure returning a set of rows.
    """
    def handle(self):

        # Get Oracle Database connection
        conn = self.out.sql['My Oracle DB']

        # Prepare a RowsOut object to receive the result set
        rows_out = RowsOut()

        # Build parameter list for the procedure
        params = [rows_out]

        # Call the procedure, populating rows
        conn.callproc('get_users', params)

        # Convert the cursor results to a list of rows
        rows = list(rows_out.get())

        # Return the list as the service response
        self.response.payload = rows

[[1,"john.doe"], [2,"jane.smith"], [3,"bob.jones"]]

# -*- coding: utf-8 -*-
from zato.server.service import Service

class GetSingleUserById(Service):
    """ # Service to fetch exactly one user or raise if not found/ambiguous.
    """
    def handle(self):

        # Get the Oracle Database connection
        conn = self.out.sql['My Oracle DB']

        # Parameterized SQL query
        query = 'select * from users where user_id = :user_id'

        # In a real service, this would be read from incoming JSON
        params = {'user_id': 1}

        # conn.one returns a dict if exactly one row, else raises (zero or multiple rows)
        result = conn.one(query, params)

        # Return the single user as the response
        self.response.payload = result

class GetSingleUserOrNoneById(Service):
    """ Service to fetch one user, None if not found, or raise an Exception if ambiguous.
    """
    def handle(self):

        # Get Oracle Database connection
        conn = self.out.sql['My Oracle DB']

        # SQL with named parameter
        query = 'select * from users where user_id = :user_id'

        # Extract user_id from payload
        params = {'user_id': 1}

        # conn.one_or_none returns a dict if one row, None if zero, raises if multiple rows
        result = conn.one_or_none(query, params)

        # Return dict or None
        self.response.payload = result