SQL Repository

Overview

A natural step in this journey, is to create a new repository, that uses the same interface that the current being used (in-memory repository), and see that everything continues working. So, let's create an SQL repository, using the SQLAlchemy package to handle the database engine stuff.

To start, we need to add the SQLAlchemy package to requirements.txt requirements.txt

@@ -1,3 +1,4 @@

 fastapi==0.61.0
 pytest==6.0.1
 requests==2.24.0
+SQLAlchemy==1.3.19

and install it:

pip install -r requirements.txt

Example

With all the necessary updates done, we continue creating a new test file:

touch tests/test_05_sqlalchemy_repository.py

First test

Let's create an initial test, to create an sql repository and add a book. First of all, we define the SQLBookRepository class, that inherits from the same BookRepository interface used by the InMemoryBooksRepository. And then, we define the initial test.

tests/test_05_sqlalchemy_repository.py

@@ -0,0 +1,32 @@

+from typing import List
+from uuid import UUID
+
+from sqlalchemy.orm import Session
+
+from tests.repositories import BookRepository
+from tests.schemas import BookInDB, Book
+
+
+class SQLBookRepository(BookRepository):
+    def __init__(self, db: Session):
+        self.db = db
+
+    def add(self, book: Book) -> BookInDB:
+        pass
+
+    def get(self, book_id: UUID) -> BookInDB:
+        pass
+
+    def list(self) -> List[BookInDB]:
+        pass
+
+
+def test_create_new_book_in_sql_repository_successfully(db_session: Session, a_book) -> None:
+    repo = SQLBookRepository(db_session)
+    created_book = repo.add(book=a_book)
+    assert created_book
+    assert isinstance(created_book, BookInDB)
+    assert created_book.title == a_book.title
+    assert created_book.author == a_book.author
+
+

Database fixtures

To work with a real SQL engine, we need to manage sessions, connections, and a bunch of things that sqlalchemy makes easier. We create a new fixture db_connection that creates a new instance of an sqlite engine in memory, and lasts for the entire test session. And then, another fixture to create a db_session for every test:

tests/test_05_sqlalchemy_repository.py

@@ -1,32 +1,48 @@

+from contextlib import closing
 from typing import List
 from uuid import UUID

-from sqlalchemy.orm import Session
+import pytest
+from sqlalchemy.engine import Connection, create_engine
+from sqlalchemy.orm import Session, sessionmaker

 from tests.repositories import BookRepository
 from tests.schemas import BookInDB, Book


 class SQLBookRepository(BookRepository):
     def __init__(self, db: Session):
         self.db = db

     def add(self, book: Book) -> BookInDB:
         pass

     def get(self, book_id: UUID) -> BookInDB:
         pass

     def list(self) -> List[BookInDB]:
         pass


+@pytest.fixture(scope="session")
+def db_connection() -> Connection:
+    db_url = "sqlite:///:memory:?check_same_thread=False"
+    engine = create_engine(db_url, pool_pre_ping=True)
+    with engine.connect() as connection:
+        yield connection
+
+
+@pytest.fixture(scope="function")
+def db_session(db_connection: Connection) -> Session:
+    session_maker = sessionmaker(autocommit=False, autoflush=False, bind=db_connection)
+    with closing(session_maker()) as session:
+        yield session
+
+
 def test_create_new_book_in_sql_repository_successfully(db_session: Session, a_book) -> None:
     repo = SQLBookRepository(db_session)
     created_book = repo.add(book=a_book)
     assert created_book
     assert isinstance(created_book, BookInDB)
     assert created_book.title == a_book.title
     assert created_book.author == a_book.author
-
-

Create Database Tables

To work with an SQL engine, we need to define tables and column types. To do this, we use the declarative method of sqlalchemy to create a new class Base, from which all data models will inherit, and which can also be used to create and drop all the database tables for each session.

tests/test_05_sqlalchemy_repository.py

@@ -1,42 +1,58 @@

 from contextlib import closing
 from typing import List
-from uuid import UUID
+from uuid import UUID, uuid4

 import pytest
+from sqlalchemy import Column, String
 from sqlalchemy.engine import Connection, create_engine
+from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy.orm import Session, sessionmaker

 from tests.repositories import BookRepository
 from tests.schemas import BookInDB, Book


 class SQLBookRepository(BookRepository):
     def __init__(self, db: Session):
         self.db = db

     def add(self, book: Book) -> BookInDB:
         pass

     def get(self, book_id: UUID) -> BookInDB:
         pass

     def list(self) -> List[BookInDB]:
         pass


+Base = declarative_base()
+
+
+class BookModel(Base):
+    __tablename__ = 'books'
+
+    # It is not recommended to store uuid as str, but for these tests is ok
+    id = Column(String, primary_key=True, index=True, default=str(uuid4()))
+    title = Column(String)
+    author = Column(String)
+
+
 @pytest.fixture(scope="session")
 def db_connection() -> Connection:
     db_url = "sqlite:///:memory:?check_same_thread=False"
     engine = create_engine(db_url, pool_pre_ping=True)
     with engine.connect() as connection:
+        Base.metadata.create_all(bind=connection)
         yield connection
+        Base.metadata.drop_all(bind=connection)


 @pytest.fixture(scope="function")
 def db_session(db_connection: Connection) -> Session:
     session_maker = sessionmaker(autocommit=False, autoflush=False, bind=db_connection)
     with closing(session_maker()) as session:
         yield session


 def test_create_new_book_in_sql_repository_successfully(db_session: Session, a_book) -> None:

Add a New Book

Next step is to write the code to add a book to the database: tests/test_05_sqlalchemy_repository.py

@@ -1,30 +1,36 @@

 from contextlib import closing
 from typing import List
 from uuid import UUID, uuid4

 import pytest
+from fastapi.encoders import jsonable_encoder
 from sqlalchemy import Column, String
 from sqlalchemy.engine import Connection, create_engine
 from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy.orm import Session, sessionmaker

 from tests.repositories import BookRepository
 from tests.schemas import BookInDB, Book


 class SQLBookRepository(BookRepository):
     def __init__(self, db: Session):
         self.db = db

     def add(self, book: Book) -> BookInDB:
-        pass
+        book_data = jsonable_encoder(book)
+        book = BookModel(**book_data)
+        self.db.add(book)
+        self.db.commit()
+        self.db.refresh(book)
+        return BookInDB(**book.__dict__)

     def get(self, book_id: UUID) -> BookInDB:
         pass

     def list(self) -> List[BookInDB]:
         pass


 Base = declarative_base()

Get a Book

Let's write a test to fetch information from the database: tests/test_05_sqlalchemy_repository.py

@@ -61,10 +61,21 @@

         yield session


 def test_create_new_book_in_sql_repository_successfully(db_session: Session, a_book) -> None:
     repo = SQLBookRepository(db_session)
     created_book = repo.add(book=a_book)
     assert created_book
     assert isinstance(created_book, BookInDB)
     assert created_book.title == a_book.title
     assert created_book.author == a_book.author
+
+
+def test_get_book_from_sql_repository_successfully(db_session: Session, a_book) -> None:
+    repo = SQLBookRepository(db_session)
+    created_book = repo.add(book=a_book)
+    book_from_db = repo.get(book_id=created_book.id)
+    assert book_from_db
+    assert isinstance(book_from_db, BookInDB)
+    assert created_book.id == book_from_db.id
+    assert a_book.title == book_from_db.title
+    assert a_book.author == book_from_db.author

And to implmemet the get method to make the test work: tests/test_05_sqlalchemy_repository.py

@@ -19,21 +19,22 @@


     def add(self, book: Book) -> BookInDB:
         book_data = jsonable_encoder(book)
         book = BookModel(**book_data)
         self.db.add(book)
         self.db.commit()
         self.db.refresh(book)
         return BookInDB(**book.__dict__)

     def get(self, book_id: UUID) -> BookInDB:
-        pass
+        book = self.db.query(BookModel).filter(BookModel.id == str(book_id)).first()
+        return BookInDB(**book.__dict__)

     def list(self) -> List[BookInDB]:
         pass


 Base = declarative_base()


 class BookModel(Base):
     __tablename__ = 'books'

Get Books List

And finally, let's do the same for a list of books: tests/test_05_sqlalchemy_repository.py

@@ -22,32 +22,33 @@

         book = BookModel(**book_data)
         self.db.add(book)
         self.db.commit()
         self.db.refresh(book)
         return BookInDB(**book.__dict__)

     def get(self, book_id: UUID) -> BookInDB:
         book = self.db.query(BookModel).filter(BookModel.id == str(book_id)).first()
         return BookInDB(**book.__dict__)

-    def list(self) -> List[BookInDB]:
-        pass
+    def list(self, skip=0, offset=5) -> List[BookInDB]:
+        books = self.db.query(BookModel).offset(skip).limit(offset).all()
+        return [BookInDB(**book.__dict__) for book in books]


 Base = declarative_base()


 class BookModel(Base):
     __tablename__ = 'books'

     # It is not recommended to store uuid as str, but for these tests is ok
-    id = Column(String, primary_key=True, index=True, default=str(uuid4()))
+    id = Column(String, primary_key=True, index=True, default=lambda x: str(uuid4()))
     title = Column(String)
     author = Column(String)


 @pytest.fixture(scope="session")
 def db_connection() -> Connection:
     db_url = "sqlite:///:memory:?check_same_thread=False"
     engine = create_engine(db_url, pool_pre_ping=True)
     with engine.connect() as connection:
         Base.metadata.create_all(bind=connection)
@@ -73,10 +74,21 @@


 def test_get_book_from_sql_repository_successfully(db_session: Session, a_book) -> None:
     repo = SQLBookRepository(db_session)
     created_book = repo.add(book=a_book)
     book_from_db = repo.get(book_id=created_book.id)
     assert book_from_db
     assert isinstance(book_from_db, BookInDB)
     assert created_book.id == book_from_db.id
     assert a_book.title == book_from_db.title
     assert a_book.author == book_from_db.author
+
+
+def test_get_list_of_books_successfully(db_session: Session) -> None:
+    repo = SQLBookRepository(db_session)
+    for book_id in range(0, 25):
+        book = Book(title=f"title {book_id}", author=f"author {book_id}")
+        repo.add(book=book)
+
+    books_from_db = repo.list()
+    assert books_from_db
+    assert len(books_from_db) == 5

Code Refactor

With all the tests going green, we can do the usual refactors to move the Book Model to a new file models.py and the SQLBookRepository class to the existing repositories.py tests/models.py

+from uuid import uuid4
+
+from sqlalchemy import Column, String
+from sqlalchemy.ext.declarative import declarative_base
+
+Base = declarative_base()
+
+
+class BookModel(Base):
+    __tablename__ = 'books'
+
+    # It is not recommended to store uuid as str, but for these tests is ok
+    id = Column(String, primary_key=True, index=True, default=lambda x: str(uuid4()))
+    title = Column(String)
+    author = Column(String)

tests/repositories.py

@@ -1,14 +1,18 @@

 from abc import ABC, abstractmethod
 from typing import List, Optional
 from uuid import UUID

+from fastapi.encoders import jsonable_encoder
+from sqlalchemy.orm import Session
+
+from tests.models import BookModel
 from tests.schemas import Book, BookInDB


 class BookRepository(ABC):

     @abstractmethod
     def add(self, book: Book) -> BookInDB:
         raise NotImplementedError

     @abstractmethod
@@ -32,10 +36,31 @@

             raise ValueError("This repository only accepts Book objects.")
         book = BookInDB(**new_book.dict())
         self.books[book.id] = book
         return book

     def get(self, book_id: UUID) -> Optional[BookInDB]:
         return self.books.get(book_id, None)

     def list(self, skip=0, offset=5) -> List[BookInDB]:
         return list(self.books.values())[skip * offset:(skip + 1) * offset]
+
+
+class SQLBookRepository(BookRepository):
+    def __init__(self, db: Session):
+        self.db = db
+
+    def add(self, book: Book) -> BookInDB:
+        book_data = jsonable_encoder(book)
+        book = BookModel(**book_data)
+        self.db.add(book)
+        self.db.commit()
+        self.db.refresh(book)
+        return BookInDB(**book.__dict__)
+
+    def get(self, book_id: UUID) -> BookInDB:
+        book = self.db.query(BookModel).filter(BookModel.id == str(book_id)).first()
+        return BookInDB(**book.__dict__)
+
+    def list(self, skip=0, offset=5) -> List[BookInDB]:
+        books = self.db.query(BookModel).offset(skip).limit(offset).all()
+        return [BookInDB(**book.__dict__) for book in books]

tests/test_05_sqlalchemy_repository.py

@@ -1,56 +1,19 @@

 from contextlib import closing
-from typing import List
-from uuid import UUID, uuid4

 import pytest
-from fastapi.encoders import jsonable_encoder
-from sqlalchemy import Column, String
 from sqlalchemy.engine import Connection, create_engine
-from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy.orm import Session, sessionmaker

-from tests.repositories import BookRepository
+from tests.models import Base
+from tests.repositories import SQLBookRepository
 from tests.schemas import BookInDB, Book
-
-
-class SQLBookRepository(BookRepository):
-    def __init__(self, db: Session):
-        self.db = db
-
-    def add(self, book: Book) -> BookInDB:
-        book_data = jsonable_encoder(book)
-        book = BookModel(**book_data)
-        self.db.add(book)
-        self.db.commit()
-        self.db.refresh(book)
-        return BookInDB(**book.__dict__)
-
-    def get(self, book_id: UUID) -> BookInDB:
-        book = self.db.query(BookModel).filter(BookModel.id == str(book_id)).first()
-        return BookInDB(**book.__dict__)
-
-    def list(self, skip=0, offset=5) -> List[BookInDB]:
-        books = self.db.query(BookModel).offset(skip).limit(offset).all()
-        return [BookInDB(**book.__dict__) for book in books]
-
-
-Base = declarative_base()
-
-
-class BookModel(Base):
-    __tablename__ = 'books'
-
-    # It is not recommended to store uuid as str, but for these tests is ok
-    id = Column(String, primary_key=True, index=True, default=lambda x: str(uuid4()))
-    title = Column(String)
-    author = Column(String)


 @pytest.fixture(scope="session")
 def db_connection() -> Connection:
     db_url = "sqlite:///:memory:?check_same_thread=False"
     engine = create_engine(db_url, pool_pre_ping=True)
     with engine.connect() as connection:
         Base.metadata.create_all(bind=connection)
         yield connection
         Base.metadata.drop_all(bind=connection)

That's all for now.