SQL and stored procedures
Put any SQL database behind a REST API - stored procedures, queries and inserts.
Integrations read from and write to databases - fetching reference data, logging audit records and calling stored procedures that encapsulate business logic. Outgoing SQL connections are configured once and used across all your services.
The pattern mirrors REST connections: get a connection by name, create a session, execute queries and map results to your response format. Sessions are closed in a finally block, so a failed query never leaks a connection.
Create a database connection
To create a connection, go to Connections > Outgoing > SQL in the Dashboard, click Create a new outgoing SQL connection and fill in the form:
- Name: CompanyDatabase
- Engine: the database type, e.g. MS SQL
- Host, port, database and username: where and how to connect
- Click OK, then click Change password in the connection's row
Execute stored procedures
# -*- coding: utf-8 -*-
# stdlib
from http import HTTPStatus
# Zato
from zato.server.service import Service
class GetCompanyGuarantors(Service):
""" Returns company guarantors from a stored procedure.
"""
name = 'companies.guarantors.list'
def handle(self) -> 'None':
# The stored procedure to call ..
stored_proc = 'EXEC company.spGetCompanyGuarantors'
# .. through the connection configured in the Dashboard.
conn = self.outgoing.sql.get('CompanyDatabase')
session = conn.session()
try:
result = session.execute(stored_proc)
# Each row maps to our own field names
guarantors = []
for row in result:
guarantors.append({
'company': row['CompanyName'],
'company_id': row['CompanyIdentifier'],
'guarantor_name': row['GuarantorName'],
'guarantor_ssn': row['GuarantorSSN'],
'guarantor_email': row['GuarantorEmail'],
'guarantor_phone': row['GuarantorPhone']
})
self.response.payload = {'guarantors': guarantors}
except Exception as e:
self.logger.error('Database error, e:`%s`', e)
self.response.status_code = HTTPStatus.INTERNAL_SERVER_ERROR
self.response.payload = {'error': str(e)}
finally:
# The session goes back to the pool no matter what happened above
session.close()
Expose the service on a channel at /api/companies/guarantors and call it:
The response is JSON with the field names your mapping chose, not the database's.
Stored procedures with parameters
Parameters are always bound, never formatted into the SQL text - the database receives the value separately from the statement, so no input can change what the statement does:
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import Service
class GetEmployeeByDepartment(Service):
name = 'employees.by-department'
input = 'department_id'
def handle(self) -> 'None':
department_id = self.request.input.department_id
# The :dept_id placeholder is bound, not interpolated
stored_proc = 'EXEC hr.spGetEmployeesByDepartment @DeptID = :dept_id'
params = {'dept_id': department_id}
conn = self.outgoing.sql.get('HRDatabase')
session = conn.session()
try:
result = session.execute(stored_proc, params)
employees = []
for row in result:
# HireDate is nullable in the database
hire_date = row['HireDate']
if hire_date is not None:
hire_date = hire_date.isoformat()
employees.append({
'id': row['EmployeeID'],
'name': row['FullName'],
'email': row['Email'],
'hire_date': hire_date
})
self.response.payload = {
'department_id': department_id,
'employees': employees,
'count': len(employees)
}
finally:
session.close()
Raw SQL queries
A query built from optional filters adds each condition and its bound parameter together, so the SQL text never contains user input:
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import Service
class SearchProducts(Service):
name = 'products.search'
input = 'query', '-category', '-min_price', '-max_price'
def handle(self) -> 'None':
query = self.request.input.query
category = self.request.input.category
min_price = self.request.input.min_price
max_price = self.request.input.max_price
# The base query with its one required filter ..
sql = """
SELECT ProductID, Name, Category, Price, InStock
FROM Products
WHERE Name LIKE :query
"""
params = {'query': f'%{query}%'}
# .. and each optional filter adds a condition with its parameter.
if category:
sql += ' AND Category = :category'
params['category'] = category
if min_price:
sql += ' AND Price >= :min_price'
params['min_price'] = min_price
if max_price:
sql += ' AND Price <= :max_price'
params['max_price'] = max_price
sql += ' ORDER BY Name'
conn = self.outgoing.sql.get('ProductDatabase')
session = conn.session()
try:
result = session.execute(sql, params)
products = []
for row in result:
products.append({
'id': row['ProductID'],
'name': row['Name'],
'category': row['Category'],
'price': float(row['Price']),
'in_stock': row['InStock']
})
self.response.payload = {'products': products}
finally:
session.close()
Insert and update operations
Writes commit on success and roll back on failure:
# -*- coding: utf-8 -*-
# stdlib
from http import HTTPStatus
# Zato
from zato.server.service import Service
class CreateAuditLog(Service):
name = 'audit.log.create'
input = 'action', 'user_id', 'details'
def handle(self) -> 'None':
action = self.request.input.action
user_id = self.request.input.user_id
details = self.request.input.details
sql = """
INSERT INTO AuditLog (Action, UserID, Details, Timestamp)
VALUES (:action, :user_id, :details, GETDATE())
"""
params = {
'action': action,
'user_id': user_id,
'details': details
}
conn = self.outgoing.sql.get('AuditDatabase')
session = conn.session()
try:
session.execute(sql, params)
session.commit()
self.response.status_code = HTTPStatus.CREATED
self.response.payload = {'status': 'logged'}
except Exception as e:
# A failed write leaves nothing half-inserted
session.rollback()
self.logger.error('Failed to create audit log, e:`%s`', e)
self.response.status_code = HTTPStatus.INTERNAL_SERVER_ERROR
self.response.payload = {'error': str(e)}
finally:
session.close()
Map results to models
Rows can populate dataclasses instead of dicts, giving the mapping named, typed fields:
# -*- coding: utf-8 -*-
# stdlib
from dataclasses import dataclass
# Zato
from zato.server.service import Service
@dataclass
class Guarantor:
company: str = ''
company_id: str = ''
name: str = ''
ssn: str = ''
email: str = ''
phone: str = ''
class GetGuarantors(Service):
name = 'guarantors.list'
def handle(self) -> 'None':
conn = self.outgoing.sql.get('CompanyDatabase')
session = conn.session()
try:
result = session.execute('EXEC company.spGetCompanyGuarantors')
guarantors = []
for row in result:
g = Guarantor()
g.company = row['CompanyName']
g.company_id = row['CompanyIdentifier']
g.name = row['GuarantorName']
g.ssn = row['GuarantorSSN']
g.email = row['GuarantorEmail']
g.phone = row['GuarantorPhone']
guarantors.append(g)
# Dataclasses become dicts for the JSON response
self.response.payload = {
'guarantors': [g.__dict__ for g in guarantors]
}
finally:
session.close()
See also
| Page | What it covers |
|---|---|
| REST channels | Exposing the database services as REST endpoints |
| Data mapping | The mapping patterns the row-to-model code follows |
| Error handling | What callers receive when a query fails |