Hey there, fellow developer! Ready to dive into the world of SQL Server API integration with Python? You're in the right place. We'll be using the powerful pyodbc package to make this happen. Buckle up, and let's get coding!
Before we jump in, make sure you've got:
pip install pyodbc
)Let's kick things off by importing our libraries and establishing a connection:
import pyodbc conn = pyodbc.connect('DRIVER={SQL Server};SERVER=your_server;DATABASE=your_db;UID=your_username;PWD=your_password') cursor = conn.cursor()
We'll use Flask for this example, but feel free to swap it out with your favorite framework:
from flask import Flask, jsonify, request app = Flask(__name__) @app.route('/') def hello(): return "Welcome to your SQL Server API!"
@app.route('/users', methods=['GET']) def get_users(): cursor.execute("SELECT * FROM users") users = cursor.fetchall() return jsonify([dict(zip([column[0] for column in cursor.description], row)) for row in users])
@app.route('/users', methods=['POST']) def add_user(): user = request.json cursor.execute("INSERT INTO users (name, email) VALUES (?, ?)", user['name'], user['email']) conn.commit() return jsonify({"message": "User added successfully"}), 201
@app.route('/users/<int:id>', methods=['PUT']) def update_user(id): user = request.json cursor.execute("UPDATE users SET name = ?, email = ? WHERE id = ?", user['name'], user['email'], id) conn.commit() return jsonify({"message": "User updated successfully"})
@app.route('/users/<int:id>', methods=['DELETE']) def delete_user(id): cursor.execute("DELETE FROM users WHERE id = ?", id) conn.commit() return jsonify({"message": "User deleted successfully"})
@app.route('/users/search') def search_users(): name = request.args.get('name') cursor.execute("SELECT * FROM users WHERE name LIKE ?", f'%{name}%') users = cursor.fetchall() return jsonify([dict(zip([column[0] for column in cursor.description], row)) for row in users])
import logging logging.basicConfig(filename='api.log', level=logging.INFO) @app.errorhandler(Exception) def handle_exception(e): logging.error(f"An error occurred: {str(e)}") return jsonify({"error": "An internal error occurred"}), 500
For connection pooling, consider using a library like SQLAlchemy. Always use parameterized queries to prevent SQL injection and improve performance.
Use environment variables for sensitive information:
import os from dotenv import load_dotenv load_dotenv() conn = pyodbc.connect(os.getenv('DB_CONNECTION_STRING'))
Here's a quick pytest example:
import pytest def test_get_users(): response = client.get('/users') assert response.status_code == 200 assert len(response.json) > 0
Consider containerizing your app with Docker:
FROM python:3.9 WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . CMD ["python", "app.py"]
And there you have it! You've just built a robust SQL Server API integration in Python. Remember, this is just the beginning. Keep exploring, keep coding, and most importantly, keep having fun with it!
For more advanced topics, check out the pyodbc documentation and dive deeper into API security best practices. The sky's the limit!
Happy coding, rockstar! 🚀