Step by Step Guide to Building a MySQL API Integration in Python

Aug 2, 2024 • 7 minute read

Introduction

Hey there, fellow developer! Ready to dive into the world of MySQL and Python? We're going to walk through building a MySQL API integration using the awesome mysql-connector-python package. This guide assumes you're already familiar with Python and APIs, so we'll keep things snappy and focus on the good stuff.

Prerequisites

Before we jump in, make sure you've got:

Python installed (I know you do, but just checking!)
A MySQL server up and running
The mysql-connector-python package (pip install mysql-connector-python)

Got all that? Great! Let's get our hands dirty.

Establishing a Database Connection

First things first, let's connect to our MySQL database:

import mysql.connector

try:
    connection = mysql.connector.connect(
        host="localhost",
        user="your_username",
        password="your_password",
        database="your_database"
    )
    print("Connected successfully!")
except mysql.connector.Error as error:
    print(f"Oops! {error}")

Easy peasy, right? Just remember to replace those placeholder credentials with your actual database info.

Creating a Database and Table

Now, let's set up our playground:

cursor = connection.cursor()

# Create database (if it doesn't exist)
cursor.execute("CREATE DATABASE IF NOT EXISTS your_database")

# Create table
cursor.execute("""
    CREATE TABLE IF NOT EXISTS users (
        id INT AUTO_INCREMENT PRIMARY KEY,
        name VARCHAR(255),
        email VARCHAR(255)
    )
""")

CRUD Operations

Time for the fun part - let's CRUD it up!

Insert

def insert_user(name, email):
    query = "INSERT INTO users (name, email) VALUES (%s, %s)"
    cursor.execute(query, (name, email))
    connection.commit()

Read

def get_users():
    cursor.execute("SELECT * FROM users")
    return cursor.fetchall()

Update

def update_user(id, name, email):
    query = "UPDATE users SET name = %s, email = %s WHERE id = %s"
    cursor.execute(query, (name, email, id))
    connection.commit()

Delete

def delete_user(id):
    query = "DELETE FROM users WHERE id = %s"
    cursor.execute(query, (id,))
    connection.commit()

Building API Endpoints

Let's use Flask to create our API endpoints:

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/users', methods=['GET', 'POST'])
def users():
    if request.method == 'GET':
        return jsonify(get_users())
    elif request.method == 'POST':
        data = request.json
        insert_user(data['name'], data['email'])
        return jsonify({"message": "User created successfully"}), 201

@app.route('/users/<int:id>', methods=['PUT', 'DELETE'])
def user(id):
    if request.method == 'PUT':
        data = request.json
        update_user(id, data['name'], data['email'])
        return jsonify({"message": "User updated successfully"})
    elif request.method == 'DELETE':
        delete_user(id)
        return jsonify({"message": "User deleted successfully"})

if __name__ == '__main__':
    app.run(debug=True)

Error Handling and Validation

Don't forget to add some error handling and input validation. Here's a quick example:

from flask import abort

def validate_user_data(data):
    if not data.get('name') or not data.get('email'):
        abort(400, description="Name and email are required")

@app.route('/users', methods=['POST'])
def create_user():
    data = request.json
    validate_user_data(data)
    try:
        insert_user(data['name'], data['email'])
        return jsonify({"message": "User created successfully"}), 201
    except mysql.connector.Error as error:
        return jsonify({"error": str(error)}), 500

Connection Pooling

For better performance, let's implement connection pooling:

from mysql.connector import pooling

connection_pool = pooling.MySQLConnectionPool(
    pool_name="mypool",
    pool_size=5,
    host="localhost",
    user="your_username",
    password="your_password",
    database="your_database"
)

def get_connection():
    return connection_pool.get_connection()

# Use this in your CRUD functions
with get_connection() as connection:
    cursor = connection.cursor()
    # Your database operations here

Security Considerations

Always use prepared statements (as we did above) to prevent SQL injection. Also, never store sensitive data like passwords in plain text - use hashing!

Testing the API

Don't forget to test your API! Here's a simple unit test example:

import unittest
import json
from your_app import app

class TestAPI(unittest.TestCase):
    def setUp(self):
        self.app = app.test_client()

    def test_create_user(self):
        response = self.app.post('/users',
                                 data=json.dumps({'name': 'Test User', 'email': '[email protected]'}),
                                 content_type='application/json')
        self.assertEqual(response.status_code, 201)

if __name__ == '__main__':
    unittest.main()

Conclusion

And there you have it! You've just built a MySQL API integration in Python. Pretty cool, right? Remember, this is just the beginning. There's always room for improvement, like adding authentication, rate limiting, or expanding your API's functionality.

Additional Resources

Want to dive deeper? Check out:

Now go forth and build amazing things! Happy coding!