sqlalchemy_different_type_of_query.py

# pre rehnquist:
# >psycopg2
# >python version 3.*
#> SQLAlchemy
#
# Outcome:
# >using Sqlalchemy create table and call data from database
# >Table sync in postgresql


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):
    """
    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'




if __name__ == '__main__':
    engine = db_connect()
    create_deals_table(engine)