Database programming (SQLite)

Beginning with SQLAlchemy Python

Ok I have been doing Python for a while now and Python is amazing. So thought of writing a blog about it in my new blog. Leaving the chitchat.

We will be performing database operations (CRUD ) in SQLite database. This is a simple database that many programming language support and recommended for beginning database programming for most of them. We will be building a simple todo application to begin with. For further practice, you can create Restaurant Menu App, Remainder App, whatever you like. Be creative. 😎😎😎

We have few steps that we will be following. Importing libraries, connecting to database (*.db) and creating session interface for performing operations.

We will be using SQLAlchemy python module to begin.

Importing dependencies

import sys
from sqlalchemy import Column, ForeignKey, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from sqlalchemy import create_engine

These are the dependencies that we will be using for begin with.

Connecting to database

base = declarative_base()
engine = create_engine('sqlite:///todo_app.db')
base.metadata.create_all(engine)

Creating Session for operations

db_session = sessionmaker(bind = engine)
session = db_session()

Ok now the basic code for importing the required dependencies,  connecting to database and opening session for other operations to the database has been completed. Now we will be looking into the actual operations Create, Read, Update and Delete operations to the database .

After completing with basic steps that we will be following.
We will be starting with building Todo Application.
We will begin with creating a file that will act as base for starting application and initializing database.

Our database will be of following structure:

download-2

db_setup.py

import sys
from sqlalchemy import Column, ForeignKey, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from sqlalchemy import create_engine

Base = declarative_base()

class Tasks(Base):
    __tablename__ = 'tasks'
    id = Column(Integer, primary_key=True, nullable=False)
    project_id = Column(Integer, ForeignKey('projects.id'))
    title = Column(String(250), nullable=False)

class Projects(Base):
    __tablename__ = 'projects'
    id = Column(Integer, primary_key=True, nullable=False)
    name  = Column(String(100), nullable=False)
    slug = Column(String(250))
    created_by = Column(String(100), nullable=False)
    created_at = Column(DateTime, nullable=False)
    updated_at = Column(DateTime, nullable=False)

engine = create_engine('sqlite:///todo_app.db')
Base.metadata.create_all(engine)

We have created two tables for the database todo_app. We create todo_app.db file as our storage file for database.
For respective tables we create each classes; Tasks and Projects. The __tablename__ variable is added to each classes that represents the table name to be used by SQLAlchemy python module.

Create new record in database

new_project = Projects(name="Project 1", slug="www.google.com")
session.add(new_project)
session.commit()

Read from database

Query from the database based on the object in python.

first_project = session.query(Projects).first()
first_project.name
first_project.slug

Read all records stored in database

all_projects = session.query(Projects).all()
for project in all_projects:
print project.name

Using filter_by query to query database

project1 = session.query(Projects).filter_by(name="Project 1")
tasks_project1 = session.query(Tasks).filter_by(project_id=project1.id)
for task in tasks_project1:
print task.title
print task.description
print task.completed
print task.slug

Update record in database

To perform update, we have to find the record to update, reset value(s), add to session and commit.

# update the project from "Project 1" to "Project 2" for the task from database
search_task = session.query(Tasks).filter_by(name="Make Tea")
new_project_id = session.query(Projects).filter_by(name="Project 2")
search_task.project_id = new_project_id.id
session.add(search_task)
session.commit()

Following is the full code for Todo application

import sys

from datetime import datetime
from sqlalchemy import Column, ForeignKey, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, sessionmaker
from sqlalchemy import create_engine

Base = declarative_base()


class Projects(Base):
    __tablename__ = 'projects'
    id = Column(Integer, primary_key=True, nullable=False)
    name = Column(String(100), nullable=False)
    slug = Column(String(250))
    created_by = Column(String(100), nullable=False)
    created_at = Column(DateTime, nullable=False)
    updated_at = Column(DateTime, nullable=False)


class Tasks(Base):
    __tablename__ = 'tasks'
    id = Column(Integer, primary_key=True, nullable=False)
    project_id = Column(Integer, ForeignKey('projects.id'))
    title = Column(String(250), nullable=False)
    description = Column(String(250), nullable=False)
    slug = Column(String(250), nullable=False)
    completed = Column(Integer)
    created_by = Column(String(100), nullable=False)
    created_at = Column(DateTime, nullable=False)
    updated_at = Column(DateTime, nullable=False)
    project = relationship(Projects)


engine = create_engine('sqlite:///todo_app.db')
Base.metadata.create_all(engine)
Base.metadata.bind = engine
DBSession = sessionmaker(bind=engine)
session = DBSession()


class Todo:
    def __init__(self):
        pass

    def create_project(self, name, slug, created_by):
        created_at = datetime.now()
        updated_at = datetime.now()

        project_name = session.query(Projects).filter_by(name=name)
        if project_name.count() == 0:
            new_project = Projects(name=name, slug=slug, created_by=created_by, created_at=created_at,
                                   updated_at=updated_at)
            session.add(new_project)
            session.commit()
            print("Project created: "+ name)
        else:
            print("Project already exist.")

    def create_tasks(self, title, description, slug, project_name, created_by):
        created_at = datetime.now()
        updated_at = datetime.now()
        project_id = session.query(Projects).filter_by(name=project_name).one()
        new_task = Tasks(title=title, description=description, slug=slug, project_id=project_id, completed=0,
                         created_by=created_by, created_at=created_at, updated_at=updated_at)
        session.add(new_task)
        session.commit()

    def read_all_project(self):
        return session.query(Projects).all()

    def read_all_tasks(self):
        return session.query(Tasks).all()

    def read_project(self, name):
        return session.query(Projects).filter_by(name=name).one()

    def read_task(self, name):
        return session.query(Tasks).filter_by(name=name).one()

    def update_project_name(self, old_project_name, new_project_name):
        search_project_name = session.query(Projects).filter_by(name=old_project_name)
        search_project_name.name = new_project_name
        session.add(search_project_name)
        session.commit()

    def update_task(self, old_task, new_task):
        search_task = session.query(Tasks).filter_by(name=old_task.id)
        if old_task.title != new_task.title:
            search_task.title = new_task.title
        elif old_task.description != new_task.description:
            search_task.description = new_task.description
        elif old_task.project_id != new_task.project_id:
            search_task.project_id = new_task.project_id
        elif old_task.slug != new_task.slug:
            search_task.slug = new_task.slug
        elif old_task.completed != new_task.completed:
            search_task.completed = new_task.completed
        elif old_task.created_by != new_task.created_by:
            search_task.created_by = new_task.created_by

        search_task.updated_at = new_task.updated_at
        session.add(search_task)
        session.commit()

    def delete_task(self, title):
        search_task = session.query(Tasks).filter_by(title=title).one()
        session.delete(search_task)
        session.commit()

    def delete_project(self, name):
        search_project = session.query(Projects).filter_by(name=name).one()
        session.delete(search_project)
        session.commit()
        print ("Project deleted: "+ name)


if __name__ == '__main__':
    todo = Todo()
    print("-----Creating a Project")
    todo.create_project("Project 1", "www.project1.com", "sagarduwal")
    print("\n")

    print("-----Reading projects")
    print("Reading multiple: ")
    for projects in todo.read_all_project():
        print projects.name
    print("\n")

    print("Reading one project: ")
    search_project = todo.read_project("Project 1")
    print search_project.name
    print("\n")

    print("-----Delete project")
    todo.delete_project("Project 1")

Leave a Comment