As part of my love letter to Python, I decided to write a simple app, to be delivered little by little, showing common concerns you would find and possible solutions you could use when writing a Python “app”. I selected a simple to-do list app because it’s a simple and common problem. I decided to start with the data access layer because I’ve recently been working with . My previous experience was full of JDBC, Hibernate, MyBatis, JPA and Spring Data, and plain drivers using NodeJS. SQLAlchemy If you ever used an , you already know that it can save a lot of time, especially when starting a project, but as the codebase grows -if it’s used indiscriminately- you can end up with an impossible-to-tune-and-maintain application. So, I decided to use over this time. ORM SQLAlchemy Core ORM You can find the codebase . I always try to follow a TDD approach, so every implementation is accompanied by its corresponding set of tests. here The actual implementation has the following components: . Defines the database schema and provides connections wrapped in a Transaction Manager that keeps the database conversational state. Schema . It’s an abstract that implements the main 4 operations of a simple CRUD. Assuming that every table uses an auto-increment id field. BaseRepo repository and . Extends BaseRepo for Users. UserRepo tests and . Extends BaseRepo for To-dos. TodoRepo tests I have added -on purpose- a relationship between Users and To-dos for testing foreign constraints. Remember that this is — kind of — a PoC from a developer that has seen more Java and JavaScript code than Python. Schema About the implementation: Uses an env variable for creating the engine Defines the DB schema Offers methods to create and drop the DB schema Offers a Context Manager (TransactionManager) that wraps the connection and handles the DB conversational state. Easy to use with Python . with statements os sqlalchemy (
    create_engine, 
    Boolean,
    Column, 
    ForeignKey,
    Integer, 
    MetaData, 
    String,
    Table
) self.engine = create_engine(os.getenv( ), echo= , future= )
        self.metadata = MetaData()
        self.tables = self.__generate_tables() TransactionManager(self) self.metadata.create_all(self.engine) self.metadata.drop_all(self.engine) { : Table( , self.metadata,
                Column( , Integer, primary_key= , autoincrement= ),
                Column( , Integer, ForeignKey( )),
                Column( , String( )),
                Column( , Boolean)
            ), : Table( , self.metadata,
                Column( , Integer, primary_key= , autoincrement= ),
                Column( , String( )),
                Column( , String( ))
            ),
        } self.schema = schema
        self.conn = self.schema.engine.connect() self self.conn.commit()
        self.conn.close() import from import : class Schema : def __init__ (self) "DB_URI" True True : def create_transaction (self) return : def create_all_tables (self) : def drop_all_tables (self) : def __generate_tables (self) return 'todo' 'todo' 'id' True True 'user_id' 'user.id' 'description' 500 'active' 'user' 'user' 'id' True True 'email' 50 'fullname' 50 : class TransactionManager : def __init__ (self, schema) : def __enter__ (self) return : def __exit__ (self, type, value, traceback) BaseRepo About the implementation: Assumes that the table is defined in the schema and has an auto-increment id field. Separates insert from update. You could implement just a save operation, similar to an upsert. Returns cursors when finding all, not an array. Useful when handling several rows. abc ABC, abstractmethod sqlalchemy insert, select, update sqlalchemy.exc InvalidRequestError self.conn = tx.conn
        self.schema = tx.schema stmt = select(self._get_table()) self.conn.execute(stmt) table = self._get_table()
        stmt = select(table).where(table.c.id == id)
        res = self.conn.execute(stmt) res: res.fetchone() table = self._get_table()
        stmt = insert(table).values(user)
        res = self.conn.execute(stmt) res.inserted_primary_key[ ] prev_user = self.find_by_id(user[ ]) prev_user: InvalidRequestError( )
        stmt = update(self._get_table()).values(user)
        self.conn.execute(stmt) from import from import from import : class BaseRepo (ABC) : def __init__ (self, tx) : def find_all (self) return : def find_by_id (self, id) if return : def insert (self, user) return 0 : def update (self, user) 'id' if not raise 'Invalid id' @abstractmethod : def _get_table (self) pass TodoRepo About the implementation: Extends BaseRepo. Implements the only required method that returns the corresponding table from the Schema. Any specific data access methods should be implemented here. base_repo BaseRepo self.schema.tables[ ] from import : class TodoRepo (BaseRepo) : def _get_table (self) return 'todo' TodoRepoTest About the implementation: Uses in-memory sqlite DB, injected via env variables. Groups in classes tests per each method in TodoRepo. Creates a new schema per test. Enables foreign constraints. pytest sqlalchemy event, insert, select sqlalchemy.exc IntegrityError, InvalidRequestError schema Schema todo_repo TodoRepo monkeypatch.setenv( , )
    schema = Schema()
    _enable_foreign_constraints(schema)
    schema.create_all_tables() schema dbapi_con.execute( )
    event.listen(schema.engine, , _fk_pragma_on_connect) schema.create_transaction() tx:
            insert_obj(tx, , { : , : , : })
            insert_obj(tx, , { : , : , : , : })
            insert_obj(tx, , { : , : , : , : })
            todos = TodoRepo(tx).find_all().fetchall() len(todos) == todos[ ][ ] == todos[ ][ ] == todos[ ][ ] == todos[ ][ ] == todos[ ][ ] == todos[ ][ ] == todos[ ][ ] == todos[ ][ ] == schema.create_transaction() tx:
            todos = TodoRepo(tx).find_all().fetchall() len(todos) == schema.create_transaction() tx:
            insert_obj(tx, , { : , : , : })
            insert_obj(tx, , { : , : , : , : })
            insert_obj(tx, , { : , : , : , : })
            todo = TodoRepo(tx).find_by_id( ) todo[ ] == todo[ ] == todo[ ] == todo[ ] == schema.create_transaction() tx:
            todo = TodoRepo(tx).find_by_id( ) todo schema.create_transaction() tx:
            insert_obj(tx, , { : , : , : })
            id = TodoRepo(tx).insert({ : , : , : })
            table = tx.schema.tables[ ]
            todos = tx.conn.execute(select(table)).fetchall() len(todos) == todos[ ][ ] == id todos[ ][ ] == todos[ ][ ] == todos[ ][ ] == schema.create_transaction() tx:
            insert_obj(tx, , { : , : , : })
            TodoRepo(tx).insert({ : , : , : , : })
            table = tx.schema.tables[ ]
            todos = tx.conn.execute(select(table)).fetchall() len(todos) == todos[ ][ ] == todos[ ][ ] == todos[ ][ ] == todos[ ][ ] == schema.create_transaction() tx, pytest.raises(IntegrityError):
            insert_obj(tx, , { : , : , : })
            TodoRepo(tx).insert({ : , : , : , : }) schema.create_transaction() tx, pytest.raises(IntegrityError):
            TodoRepo(tx).insert({ : , : , : , : }) schema.create_transaction() tx, pytest.raises(IntegrityError):
            repo = TodoRepo(tx)
            repo.insert({ : , : , : , : })
            repo.insert({ : , : , : , : }) schema.create_transaction() tx:
            insert_obj(tx, , { : , : , : })
            insert_obj(tx, , { : , : , : })
            insert_obj(tx, , { : , : , : , : })
            TodoRepo(tx).update({ : , : , : , : })
            table = tx.schema.tables[ ]
            todos = tx.conn.execute(select(table)).fetchall() len(todos) == todos[ ][ ] == todos[ ][ ] == todos[ ][ ] == todos[ ][ ] == schema.create_transaction() tx, pytest.raises(InvalidRequestError):
            TodoRepo(tx).update({ : , : , : , : }) table = tx.schema.tables[table_name]
    stmt = insert(table).values(obj)
    tx.conn.execute(stmt) import from import from import from import from import @pytest.fixture : def schema (monkeypatch) "DB_URI" "sqlite://" return : def _enable_foreign_constraints (schema) : def _fk_pragma_on_connect (dbapi_con, con_record) 'pragma foreign_keys=ON' 'connect' : class TestTodoRepo_find_all : def test_when_multiple_todos (self, schema) with as 'user' 'id' 1 'email' 'a@test.com' 'fullname' 'fullname a' 'todo' 'id' 1 'user_id' 1 'description' 'description a' 'active' True 'todo' 'id' 2 'user_id' 1 'description' 'description b' 'active' False assert 2 assert 0 'id' 1 assert 0 'user_id' 1 assert 0 'description' 'description a' assert 0 'active' True assert 1 'id' 2 assert 1 'user_id' 1 assert 1 'description' 'description b' assert 1 'active' False : def test_when_empty (self, schema) with as assert 0 : class TestTodoRepo_find_by_id : def test_when_todo_exists (self, schema) with as 'user' 'id' 1 'email' 'a@test.com' 'fullname' 'fullname a' 'todo' 'id' 1 'user_id' 1 'description' 'description a' 'active' True 'todo' 'id' 2 'user_id' 1 'description' 'description b' 'active' False 1 assert 'id' 1 assert 'user_id' 1 assert 'description' 'description a' assert 'active' True : def test_when_todo_does_not_exists (self, schema) with as 1 assert is None : class TestTodoRepo_insert : def test_when_auto_increment_id (self, schema) with as 'user' 'id' 1 'email' 'a@test.com' 'fullname' 'fullname a' 'user_id' 1 'description' 'description a' 'active' True 'todo' assert 1 assert 0 'id' assert 0 'user_id' 1 assert 0 'description' 'description a' assert 0 'active' True : def test_when_set_id (self, schema) with as 'user' 'id' 1 'email' 'a@test.com' 'fullname' 'fullname a' 'id' 99 'user_id' 1 'description' 'description a' 'active' True 'todo' assert 1 assert 0 'id' 99 assert 0 'user_id' 1 assert 0 'description' 'description a' assert 0 'active' True : def test_when_invalid_id (self, schema) with as 'user' 'id' 1 'email' 'a@test.com' 'fullname' 'fullname a' 'id' 'invalid' 'user_id' 1 'description' 'description a' 'active' True : def test_when_invalid_user_id (self, schema) with as 'id' 1 'user_id' 1 'description' 'description a' 'active' True : def test_when_duplicated_id (self, schema) with as 'id' 1 'user_id' 1 'description' 'description a' 'active' True 'id' 1 'user_id' 1 'description' 'description a' 'active' True : class TestTodoRepo_update : def test_when_todo_exists (self, schema) with as 'user' 'id' 1 'email' 'a@test.com' 'fullname' 'fullname a' 'user' 'id' 2 'email' 'b@test.com' 'fullname' 'fullname b' 'todo' 'id' 1 'user_id' 1 'description' 'description a' 'active' True 'id' 1 'user_id' 2 'description' 'description b' 'active' False 'todo' assert 1 assert 0 'id' 1 assert 0 'user_id' 2 assert 0 'description' 'description b' assert 0 'active' False : def test_when_todo_does_not_exists (self, schema) with as 'id' 1 'user_id' 1 'description' 'description b' 'active' False : def insert_obj (tx, table_name, obj) Final words Definitively, SQLAlchemy is super powerful and simple at the same time! Really impressed by how easy it was to write these simple repos. The Java developer inside me says: change this and use objects! Maybe next time I’ll try the ORM implementation. However, my JavaScript developer side says: hell yeah! Forget about classes and open yourself to dictionaries. I don’t have a strong opinion, but I do know that you can write the same horrible app with and without classes. Still not sure about the performance of SQLAlchemy, what I read out there is that it’s pretty decent. Again, it’s going to significantly change depending on the queries you write! Hope you find useful this post. My plan is to continue expanding this project to address different concerns and possible solutions. Thanks for reading!

Super

Python Cheatsheet: Mock Object Library

Nominated for 2022 - HackerNoon Contributor of the Year - Testing

Nominated for 2022 - Pythonistas Paradise

Too Long; Didn't Read

Make resilience your competitive advantage

Building a To-do List App with Python: Data Access Layer with SQLAlchemy

About Author

Comments

TOPICS

THIS ARTICLE WAS FEATURED IN

Related Stories

Untitled Story

Python Cheatsheet: Mock Object Library

What is Python Good for? Why Beginner Should Learn Python?

11 Tips And Tricks To Write Better Python Code

199 Stories To Learn About Python Tutorials

15 Books for Computer Science Students

Python Cheatsheet: Mock Object Library

What is Python Good for? Why Beginner Should Learn Python?

11 Tips And Tricks To Write Better Python Code

199 Stories To Learn About Python Tutorials

15 Books for Computer Science Students

Light-Mode

Classic

Newspaper

Dark-Mode

Neon Noir

Minty

HN StartUps