Prerequisites
- Basic understanding of programming concepts 📝
- Python installation (3.8+) 🐍
- VS Code or preferred IDE 💻
What you'll learn
- Understand the concept fundamentals 🎯
- Apply the concept in real projects 🏗️
- Debug common issues 🐛
- Write clean, Pythonic code ✨
🎯 Introduction
Welcome to this exciting tutorial on stored procedures in Python! 🎉 Have you ever wondered how to supercharge your database operations while keeping your business logic secure and efficient? That’s exactly what stored procedures can do for you!
In this guide, we’ll explore how to call stored procedures from Python, turning your database into a powerful ally. Whether you’re building e-commerce platforms 🛒, financial systems 💰, or data processing pipelines 📊, mastering stored procedures will level up your database game!
By the end of this tutorial, you’ll be confidently calling stored procedures like a database wizard! Let’s dive in! 🏊♂️
📚 Understanding Stored Procedures
🤔 What are Stored Procedures?
Think of stored procedures as pre-written recipes 🍳 stored in your database. Just like how a restaurant chef prepares signature dishes using tested recipes, stored procedures are pre-compiled SQL scripts that perform specific database operations.
In Python terms, stored procedures are like functions that live inside your database. This means you can:
- ✨ Execute complex operations with a single call
- 🚀 Improve performance with pre-compiled SQL
- 🛡️ Enhance security by limiting direct table access
- 📦 Package business logic in the database
💡 Why Use Stored Procedures?
Here’s why developers love stored procedures:
- Performance Boost 🚀: Pre-compiled and optimized by the database
- Security Shield 🛡️: Prevent SQL injection and control access
- Network Efficiency 📡: Reduce data transfer between app and database
- Code Reusability 🔄: Share logic across multiple applications
Real-world example: Imagine an online banking system 🏦. With stored procedures, you can handle complex transactions (checking balance, validating limits, transferring money) in one secure, atomic operation!
🔧 Basic Syntax and Usage
📝 Simple Example with MySQL
Let’s start with a friendly example using MySQL and PyMySQL:
# 👋 Hello, Stored Procedures!
import pymysql
# 🎨 Connect to database
connection = pymysql.connect(
host='localhost',
user='your_user',
password='your_password',
database='your_database'
)
try:
with connection.cursor() as cursor:
# 🎯 Call a simple stored procedure
cursor.callproc('get_user_count')
# 📊 Fetch the result
result = cursor.fetchone()
print(f"Total users: {result[0]} 🎉")
finally:
connection.close()💡 Explanation: Notice how we use callproc() to execute the stored procedure. It’s that simple! The procedure runs on the database server and returns results.
🎯 Common Patterns
Here are patterns you’ll use daily:
# 🏗️ Pattern 1: Calling with parameters
def get_user_orders(user_id):
with connection.cursor() as cursor:
# 📦 Pass parameters to stored procedure
cursor.callproc('get_orders_by_user', [user_id])
# 🎨 Fetch all results
orders = cursor.fetchall()
return orders
# 🎨 Pattern 2: Getting output parameters
def calculate_discount(order_total):
with connection.cursor() as cursor:
# 🔄 Call procedure with IN and OUT parameters
args = [order_total, 0] # Second param will receive output
cursor.callproc('calculate_customer_discount', args)
# 💰 Get the output parameter
cursor.execute('SELECT @_calculate_customer_discount_1')
discount = cursor.fetchone()[0]
return discount
# 🔄 Pattern 3: Handling multiple result sets
def get_dashboard_data():
with connection.cursor() as cursor:
# 🚀 Call procedure that returns multiple datasets
cursor.callproc('get_dashboard_stats')
# 📊 Process multiple result sets
daily_sales = cursor.fetchall()
cursor.nextset()
top_products = cursor.fetchall()
cursor.nextset()
active_users = cursor.fetchall()
return daily_sales, top_products, active_users💡 Practical Examples
🛒 Example 1: E-Commerce Order Processing
Let’s build something real - an order processing system:
# 🛍️ E-commerce order processor
import pymysql
from datetime import datetime
from decimal import Decimal
class OrderProcessor:
def __init__(self, db_config):
self.db_config = db_config
# 🛒 Process a new order
def process_order(self, user_id, cart_items):
connection = pymysql.connect(**self.db_config)
try:
with connection.cursor() as cursor:
# 💳 Call order processing stored procedure
order_id = 0
total_amount = Decimal('0.00')
status = ''
# 📦 Prepare parameters
args = [
user_id, # IN: user ID
str(cart_items), # IN: cart items as JSON
order_id, # OUT: new order ID
total_amount, # OUT: total amount
status # OUT: processing status
]
# 🚀 Execute the stored procedure
cursor.callproc('process_new_order', args)
# 📊 Get output parameters
cursor.execute(
'SELECT @_process_new_order_2, ' +
'@_process_new_order_3, @_process_new_order_4'
)
result = cursor.fetchone()
order_id = result[0]
total_amount = result[1]
status = result[2]
# ✅ Commit the transaction
connection.commit()
print(f"🎉 Order #{order_id} created!")
print(f"💰 Total: ${total_amount}")
print(f"📋 Status: {status}")
return {
'order_id': order_id,
'total': total_amount,
'status': status
}
except Exception as e:
# ❌ Rollback on error
connection.rollback()
print(f"😱 Order processing failed: {e}")
raise
finally:
connection.close()
# 🎮 Let's use it!
processor = OrderProcessor({
'host': 'localhost',
'user': 'shop_user',
'password': 'secure_pass',
'database': 'ecommerce_db'
})
# 🛒 Sample cart
cart = [
{'product_id': 101, 'quantity': 2, 'price': 29.99},
{'product_id': 205, 'quantity': 1, 'price': 49.99}
]
order = processor.process_order(user_id=12345, cart_items=cart)🎯 Try it yourself: Add a method to cancel orders and apply discount codes!
🎮 Example 2: Game Leaderboard System
Let’s make it fun with a gaming leaderboard:
# 🏆 Game leaderboard manager
import psycopg2 # Using PostgreSQL for this example
from contextlib import contextmanager
class GameLeaderboard:
def __init__(self, db_url):
self.db_url = db_url
@contextmanager
def get_db_cursor(self):
# 🔄 Database connection manager
conn = psycopg2.connect(self.db_url)
try:
with conn.cursor() as cursor:
yield cursor
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()
# 🎯 Submit a new score
def submit_score(self, player_name, score, level):
with self.get_db_cursor() as cursor:
# 🚀 Call score submission procedure
cursor.callproc('submit_player_score', [
player_name, # 👤 Player name
score, # 🎯 Score achieved
level # 📈 Level completed
])
# 📊 Get player's rank
result = cursor.fetchone()
if result:
rank = result[0]
print(f"🎉 {player_name} is now rank #{rank}!")
# 🏆 Check for achievements
cursor.callproc('check_achievements', [player_name])
achievements = cursor.fetchall()
if achievements:
print("🌟 New achievements unlocked:")
for achievement in achievements:
print(f" 🏆 {achievement[0]}")
return rank
# 📋 Get top players
def get_leaderboard(self, limit=10):
with self.get_db_cursor() as cursor:
# 🏅 Call leaderboard procedure
cursor.callproc('get_top_players', [limit])
leaderboard = cursor.fetchall()
print("🏆 LEADERBOARD 🏆")
print("-" * 40)
for rank, (name, score, level) in enumerate(leaderboard, 1):
emoji = "🥇" if rank == 1 else "🥈" if rank == 2 else "🥉" if rank == 3 else "🎮"
print(f"{emoji} #{rank}: {name} - Score: {score} (Level {level})")
return leaderboard
# 📊 Get player statistics
def get_player_stats(self, player_name):
with self.get_db_cursor() as cursor:
# 📈 Call stats procedure
cursor.callproc('get_player_statistics', [player_name])
stats = cursor.fetchone()
if stats:
return {
'total_games': stats[0],
'high_score': stats[1],
'avg_score': stats[2],
'favorite_level': stats[3],
'play_time': stats[4]
}
# 🎮 Test the leaderboard!
leaderboard = GameLeaderboard('postgresql://user:pass@localhost/gamedb')
# 🚀 Submit some scores
leaderboard.submit_score("SpeedyGonzales", 9500, 12)
leaderboard.submit_score("DragonSlayer", 8700, 10)
leaderboard.submit_score("PixelNinja", 10200, 15)
# 🏆 Show the leaderboard
leaderboard.get_leaderboard()🚀 Advanced Concepts
🧙♂️ Advanced Topic 1: Handling Complex Data Types
When you’re ready to level up, work with complex return types:
# 🎯 Advanced stored procedure handling
import json
import cx_Oracle # Oracle example
class AdvancedProcedureHandler:
def __init__(self, connection_string):
self.connection_string = connection_string
# 🪄 Handle cursor output from procedures
def get_complex_report(self, report_type, date_range):
connection = cx_Oracle.connect(self.connection_string)
try:
cursor = connection.cursor()
# 📊 Create cursor for output
report_cursor = connection.cursor()
# 🚀 Call procedure that returns a cursor
cursor.callproc('generate_complex_report', [
report_type,
date_range['start'],
date_range['end'],
report_cursor # OUT parameter as cursor
])
# 🎨 Process the cursor results
columns = [col[0] for col in report_cursor.description]
results = []
for row in report_cursor:
# ✨ Convert to dictionary
results.append(dict(zip(columns, row)))
return results
finally:
connection.close()
# 🏗️ Handle JSON returns
def process_json_procedure(self, operation, data):
connection = cx_Oracle.connect(self.connection_string)
try:
cursor = connection.cursor()
# 📦 Prepare JSON data
json_input = json.dumps(data)
json_output = cursor.var(cx_Oracle.STRING)
# 💫 Call JSON processing procedure
cursor.callproc('process_json_data', [
operation,
json_input,
json_output
])
# 🎯 Parse JSON result
result = json.loads(json_output.getvalue())
return result
finally:
connection.close()🏗️ Advanced Topic 2: Async Stored Procedures
For the brave developers using async Python:
# 🚀 Async stored procedure handling
import asyncpg # PostgreSQL async driver
import asyncio
class AsyncProcedureManager:
def __init__(self, db_url):
self.db_url = db_url
self.pool = None
async def initialize(self):
# 🌟 Create connection pool
self.pool = await asyncpg.create_pool(self.db_url)
async def close(self):
# 🔄 Close connection pool
await self.pool.close()
# ⚡ Async procedure call
async def call_procedure_async(self, proc_name, *args):
async with self.pool.acquire() as connection:
# 🎯 Call stored procedure asynchronously
result = await connection.fetch(
f"SELECT * FROM {proc_name}($1, $2, $3)",
*args
)
return result
# 🏃♂️ Batch processing with procedures
async def batch_process_orders(self, order_ids):
tasks = []
# 📦 Create tasks for parallel execution
for order_id in order_ids:
task = self.call_procedure_async(
'process_order_async',
order_id,
'PROCESSING',
asyncio.get_event_loop().time()
)
tasks.append(task)
# 🚀 Execute all procedures concurrently
results = await asyncio.gather(*tasks)
print(f"✨ Processed {len(results)} orders in parallel!")
return results
# 🎮 Test async procedures
async def main():
manager = AsyncProcedureManager('postgresql://user:pass@localhost/db')
await manager.initialize()
# 🚀 Process multiple orders concurrently
order_ids = [1001, 1002, 1003, 1004, 1005]
results = await manager.batch_process_orders(order_ids)
await manager.close()
# 🏃♂️ Run it!
# asyncio.run(main())⚠️ Common Pitfalls and Solutions
😱 Pitfall 1: Parameter Mismatch
# ❌ Wrong way - incorrect parameter count!
cursor.callproc('update_user_profile', ['John']) # 💥 Procedure expects 3 params!
# ✅ Correct way - match all parameters!
cursor.callproc('update_user_profile', [
'John', # username
'[email protected]', # email
25 # age
])🤯 Pitfall 2: Forgetting to Handle Output Parameters
# ❌ Dangerous - losing output values!
def calculate_tax(amount):
cursor.callproc('calc_tax', [amount, 0])
# Output parameter is lost! 😰
# ✅ Safe - properly retrieve output!
def calculate_tax(amount):
tax_amount = 0
cursor.callproc('calc_tax', [amount, tax_amount])
# 📊 Retrieve the output parameter
cursor.execute('SELECT @_calc_tax_1')
tax = cursor.fetchone()[0]
return tax # ✅ Got it!🐛 Pitfall 3: Not Handling Multiple Result Sets
# ❌ Missing data - only gets first result set!
cursor.callproc('get_full_report')
data = cursor.fetchall() # 💥 Missing other result sets!
# ✅ Complete - get all result sets!
cursor.callproc('get_full_report')
results = []
while True:
data = cursor.fetchall()
results.append(data)
if not cursor.nextset(): # 🔄 Check for more sets
break
print(f"📊 Retrieved {len(results)} result sets!")🛠️ Best Practices
- 🎯 Use Connection Pooling: Don’t create new connections for each call
- 📝 Document Parameters: Clearly specify IN, OUT, and INOUT parameters
- 🛡️ Handle Exceptions: Always use try-except blocks with rollback
- 🎨 Use Prepared Statements: Combine with procedures for extra security
- ✨ Keep It Simple: Don’t put entire application logic in procedures
🧪 Hands-On Exercise
🎯 Challenge: Build a Banking Transaction System
Create a secure banking system with stored procedures:
📋 Requirements:
- ✅ Transfer money between accounts with validation
- 🏷️ Check account balance before transfers
- 👤 Log all transactions for audit
- 📅 Generate monthly statements
- 🎨 Each transaction needs a unique reference!
🚀 Bonus Points:
- Add fraud detection
- Implement transaction limits
- Create account interest calculations
💡 Solution
🔍 Click to see solution
# 🎯 Secure banking transaction system!
import pymysql
from datetime import datetime
from decimal import Decimal
import uuid
class BankingSystem:
def __init__(self, db_config):
self.db_config = db_config
# 💰 Transfer money between accounts
def transfer_money(self, from_account, to_account, amount):
connection = pymysql.connect(**self.db_config)
try:
with connection.cursor() as cursor:
# 🎯 Generate unique transaction reference
transaction_ref = f"TXN-{uuid.uuid4().hex[:8].upper()}"
# 📊 Call transfer stored procedure
success = False
message = ''
new_balance = Decimal('0.00')
args = [
from_account, # IN: source account
to_account, # IN: destination account
amount, # IN: transfer amount
transaction_ref, # IN: unique reference
success, # OUT: success flag
message, # OUT: status message
new_balance # OUT: new balance
]
cursor.callproc('transfer_funds', args)
# 📈 Get output parameters
cursor.execute(
'SELECT @_transfer_funds_4, ' +
'@_transfer_funds_5, @_transfer_funds_6'
)
result = cursor.fetchone()
success = result[0]
message = result[1]
new_balance = result[2]
if success:
connection.commit()
print(f"✅ Transfer successful! Ref: {transaction_ref}")
print(f"💰 New balance: ${new_balance}")
else:
connection.rollback()
print(f"❌ Transfer failed: {message}")
return {
'success': success,
'reference': transaction_ref,
'message': message,
'new_balance': new_balance
}
except Exception as e:
connection.rollback()
print(f"😱 Transaction error: {e}")
raise
finally:
connection.close()
# 📊 Check account balance
def check_balance(self, account_number):
connection = pymysql.connect(**self.db_config)
try:
with connection.cursor() as cursor:
cursor.callproc('get_account_balance', [account_number])
result = cursor.fetchone()
if result:
balance = result[0]
print(f"💳 Account {account_number}")
print(f"💰 Balance: ${balance}")
return balance
finally:
connection.close()
# 📋 Generate statement
def generate_statement(self, account_number, month, year):
connection = pymysql.connect(**self.db_config)
try:
with connection.cursor() as cursor:
# 📅 Call statement generation procedure
cursor.callproc('generate_monthly_statement', [
account_number,
month,
year
])
# 📊 Get statement data
transactions = cursor.fetchall()
print(f"\n📋 STATEMENT FOR {month}/{year}")
print("=" * 50)
total_debit = Decimal('0.00')
total_credit = Decimal('0.00')
for txn in transactions:
date, desc, amount, type_ = txn
emoji = "➖" if type_ == 'DEBIT' else "➕"
print(f"{emoji} {date}: {desc} - ${amount}")
if type_ == 'DEBIT':
total_debit += amount
else:
total_credit += amount
print("=" * 50)
print(f"📊 Total Credits: ${total_credit}")
print(f"📊 Total Debits: ${total_debit}")
print(f"💰 Net Change: ${total_credit - total_debit}")
return transactions
finally:
connection.close()
# 🎮 Test it out!
bank = BankingSystem({
'host': 'localhost',
'user': 'bank_user',
'password': 'secure_password',
'database': 'banking_db'
})
# 💸 Make a transfer
bank.transfer_money('ACC-12345', 'ACC-67890', Decimal('100.00'))
# 💰 Check balance
bank.check_balance('ACC-12345')
# 📋 Generate statement
bank.generate_statement('ACC-12345', 3, 2024)🎓 Key Takeaways
You’ve learned so much! Here’s what you can now do:
- ✅ Call stored procedures from Python with confidence 💪
- ✅ Handle parameters (IN, OUT, INOUT) like a pro 🛡️
- ✅ Process multiple result sets efficiently 🎯
- ✅ Implement secure database operations with procedures 🔒
- ✅ Build high-performance database applications 🚀
Remember: Stored procedures are powerful tools that can make your database operations faster, safer, and more maintainable! 🤝
🤝 Next Steps
Congratulations! 🎉 You’ve mastered calling stored procedures from Python!
Here’s what to do next:
- 💻 Practice with the banking system exercise above
- 🏗️ Create stored procedures for your existing projects
- 📚 Learn about database-specific features (triggers, functions)
- 🌟 Explore async database operations for high-performance apps
Remember: Every database expert started by calling their first stored procedure. Keep practicing, keep learning, and most importantly, have fun building amazing database-driven applications! 🚀
Happy coding! 🎉🚀✨