Connecting PostgreSQL with python sqlalchemy orm.

In this tutorial I will show you how to create a CRUD script by using python sqlalchemy orm.python is a multipurpose and most popular language in the world. when writing a python script, we need to store data in a different database. Sqlalchemy, best orm for python, can help us to connect with a different type of SQL database however I this tutorial I will show you how to create CRUD script by using PostgreSQL database.

Let’s assume you already know python virtual environment and you already install in your pc.

You need to install the following library.

pip install sqlalchemy
pip install psycopg2

If you face any problem to install psycopg2 in your pc, you can try for this

pip install psycopg2-binary

you have successfully installed library in your pc now we can start writing the script

from sqlalchemy import (
    Table,
    Column,
    Index,
    Integer,
    Text,
    String,
    DateTime,
    Date,
    ForeignKey,
    create_engine,
    desc,
    asc,
    Boolean,
    and_
)
from sqlalchemy.orm import load_only
# from sqlalchemy import create_engine, Column, Integer, String, DateTime,Text, DATE, Boolean, Table, ForeignKey, TIMESTAMP
from sqlalchemy.dialects.postgresql import ARRAY, UUID
from sqlalchemy.orm import relationship, backref
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.engine.url import URL
import datetime
from sqlalchemy.orm import sessionmaker
DeclarativeBase = declarative_base()
DATABASE = {
    'drivername': 'postgres',
    'host': '127.0.0.1',
    'port': '5432',
    'username': 'admin',
    'password': '789',
    'database': 'tesdb'
}
def db_connect():
    """
    Performs database connection using database settings from settings.py.
    Returns sqlalchemy engine instance
    """
    return create_engine(URL(**DATABASE))
def create_deals_table(engine):
    """"""
    DeclarativeBase.metadata.create_all(engine)
def db_session():
    engine = db_connect()
    Session = sessionmaker(bind=engine)
    session = Session()
    return session
DBSession = db_session()
class Topic(DeclarativeBase):
    """
    define a table name of topic
    """
    __tablename__ = "mymodel"
    id = Column(Integer, primary_key=True)
    title = Column(String(36))
    description = Column(String(36))
    created_by = Column(String(36))
    created_on = Column(DateTime)
    is_published = Column(Boolean, default=False)
Now connect with database and crate table

# function calling
if __name__ == '__main__':
    engine = db_connect()
    create_deals_table(engine)

Define crude functionality for Topic table

class Topic(DeclarativeBase):
    """
    opinion of a comment store in this table.
    """
    __tablename__ = "mymodel"
    id = Column(Integer, primary_key=True)
    title = Column(String(36))
    description = Column(String(36))
    created_by = Column(String(36))
    created_on = Column(DateTime)
    is_published = Column(Boolean, default=False)
    # get all item from a table
    @classmethod
    def by_all(cls):
        query = DBSession.query(Topic).all()
        query = DBSession.query(Topic).order_by(desc(Topic.created_on)).limit(10)
        return query
     # get all with descending order and limit from a table
    @classmethod
    def by_all_limit(cls):
        query = DBSession.query(Topic).order_by(desc(Topic.created_on)).limit(10)
        return query
    # get all with multiple filter and multiple field option
    @classmethod
    def by_all_filter(cls, created_by):
        query = DBSession.query(Topic).filter(and_((Topic.created_by == created_by), (Topic.is_published == True))).options(load_only( "title", "description", "created_by", "created_on")).order_by(desc(Topic.content_timestamp)).limit(10)
        return query
    # get single item call by id
    @classmethod
    def by_id(cls, id):
        query = DBSession.query(Topic).filter_by(id=id).first()
        return query
    # update single topic by id
    @classmethod
    def update_topic(cls, topic_id, **kwargs):
        DBSession.query(Topic).filter_by(id=topic_id).update(kwargs)
        DBSession.commit()
        return 'topic updated'
    # delete single topic by id
    @classmethod
    def delete_topic(cls, topic_id):
        DBSession.query(Topic).filter_by(topic_id=topic_id).delete()
        DBSession.commit()
        return 'topic deteted'
    # create topic by item
    @classmethod
    def create_topic(cls, **kwargs):
        api = Topic(**kwargs)
        DBSession.add(api)
        DBSession.commit()
        return 'topic created'

You can call those functions in your script, wherever you want.