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.
