Database: MySQL

SQLite here is for demonstration only. If you want a more production-ready database, consider using MySQL or mongoDB.

Switch to MySQL

Install MySQL on your machine and start it.

Note: Remember to create indexes on tables based on your need in a production project.

Add mysql dependency:

pip3 install mysql-connector-python

It’s the official MySQL driver written in Python.

Update requirements.txt:

pip3 freeze > requirements.txt

Update code.

Add infrastructure/database/mysql.py:

import mysql.connector
from typing import Any, List, Optional

from books.infrastructure.config import DBConfig

from ...domain.gateway import BookManager
from ...domain.model import Book


class MySQLPersistence(BookManager):
    def __init__(self, c: DBConfig):
        self.conn = mysql.connector.connect(
            host=c.host,
            port=c.port,
            user=c.user,
            password=c.password,
            database=c.database,
            autocommit=True
        )
        self.cursor = self.conn.cursor(dictionary=True)
        self._create_table()

    def _create_table(self):
        self.cursor.execute('''
            CREATE TABLE IF NOT EXISTS books (
            id INT AUTO_INCREMENT PRIMARY KEY,
            title VARCHAR(255) NOT NULL,
            author VARCHAR(255) NOT NULL,
            published_at DATE NOT NULL,
            description TEXT NOT NULL,
            isbn VARCHAR(15) NOT NULL,
            total_pages INT NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
        );
        ''')

    def create_book(self, b: Book) -> int:
        self.cursor.execute('''
            INSERT INTO books (title, author, published_at, description, isbn, total_pages) VALUES (%s, %s, %s, %s, %s, %s)
        ''', (b.title, b.author, b.published_at, b.description, b.isbn, b.total_pages))
        return self.cursor.lastrowid or 0

    def update_book(self, id: int, b: Book) -> None:
        self.cursor.execute('''
            UPDATE books SET title=%s, author=%s, published_at=%s, description=%s, isbn=%s, total_pages=%s WHERE id=%s
        ''', (b.title, b.author, b.published_at, b.description, b.isbn, b.total_pages, id))

    def delete_book(self, id: int) -> None:
        self.cursor.execute('''
            DELETE FROM books WHERE id=%s
        ''', (id,))

    def get_book(self, id: int) -> Optional[Book]:
        self.cursor.execute('''
            SELECT * FROM books WHERE id=%s
        ''', (id,))
        result: Any = self.cursor.fetchone()
        if result is None:
            return None
        return Book(**result)

    def get_books(self) -> List[Book]:
        self.cursor.execute('''
            SELECT * FROM books
        ''')
        results: List[Any] = self.cursor.fetchall()
        return [Book(**result) for result in results]

Notice that the key difference is that we use mysql.connector to replace sqlite3.

Tune infrastructure/database/init.py:

@@ -1 +1,2 @@
 from .sqlite import SQLitePersistence
+from .mysql import MySQLPersistence

Add mysql connection config item into the DBConfig struct in infrastructure/config/config.py:

@@ -4,6 +4,11 @@ from dataclasses import dataclass
 @dataclass
 class DBConfig:
     file_name: str
+    host: str
+    port: int
+    user: str
+    password: str
+    database: str

Update WireHelper to switch the dependency in application/wire_helper.py:

@@ -1,15 +1,15 @@
 from ..domain.gateway import BookManager
 from ..infrastructure.config import Config
-from ..infrastructure.database import SQLitePersistence
+from ..infrastructure.database import MySQLPersistence
 
 
 class WireHelper:
-    def __init__(self, persistence: SQLitePersistence):
+    def __init__(self, persistence: MySQLPersistence):
         self.persistence = persistence
 
     @classmethod
     def new(cls, c: Config):
-        db = SQLitePersistence(c.db.file_name)
+        db = MySQLPersistence(c.db)
         return cls(db)
 
     def book_manager(self) -> BookManager:

Put in mysql connection parammeters in main.py:

@@ -9,7 +9,12 @@ c = Config(
         8080
     ),
     DBConfig(
-        "test.db"
+        "test.db",
+        "127.0.0.1",
+        3306,
+        "test_user",
+        "test_pass",
+        "lr_book"
     )
 )
 wire_helper = WireHelper.new(c)

Voila! Your api server is powered by MySQL now!

❮ Prev Next ❯