from sqlalchemy import create_engine,text
from sqlalchemy.orm import sessionmaker, declarative_base
from datetime import datetime
from sqlalchemy.orm import relationship
from sqlalchemy import Column, ForeignKey, Integer, String, DateTime, Text,
Boolean
from sqlalchemy.orm import scoped_session
from sqlalchemy.pool import QueuePool
import psycopg2
import sqlalchemy
# conn = psycopg2.connect(database="masterdb",
# user="kft_user",
# password="6dbvghdjhh78bs",
#
host="db-primary.crszysz9bdut.ap-south-1.rds.amazonaws.com",
# port="5432")
conn = psycopg2.connect("db_url")
cursor = conn.cursor()
SQLALCHEMY_DATABASE_URL = """db_url"""
dbschema = 'kft_dev_db'
engine = create_engine(SQLALCHEMY_DATABASE_URL, poolclass=QueuePool,
pool_size=5, max_overflow=10)
connectionpool = engine.connect()
con = engine.connect()
Base = declarative_base()
session_factory = sessionmaker(autocommit=False, autoflush=False,bind=engine
)
Session = scoped_session(session_factory)
# print('Global Session', Session)
#--------------------------------------------
#models
class User(Base):
__tablename__ = "org_users"
__table_args__ = {'schema': dbschema}
org_user_id= Column(Integer, primary_key=True, index=True)
org_role_id_ref=Column(Integer,ForeignKey(f'{dbschema}
.org_roles.org_role_id', ondelete='CASCADE'))
first_name = Column (String(128), nullable=False)
last_name = Column(String(128), nullable=False)
email = Column(String(128), nullable=False)
auth_token = Column(Text, default=None)
encrypted_password = Column(Text, nullable=False)
mobile_number = Column(String(128))
designation = Column(String(128))
is_verified = Column(Boolean, default=False)
is_active = Column(Boolean, default=True)
created_at = Column(DateTime, default=datetime.utcnow(), nullable=False)
updated_at = Column(DateTime, default=None)
roles = relationship('Roles', back_populates='user_reg')
class Roles(Base):
__tablename__ = "org_roles"
__table_args__={'schema': dbschema}
org_role_id = Column(Integer, primary_key = True, index=True)
role_name = Column(String(128), nullable=False)
is_active = Column(Boolean, default=True)
created_at = Column(DateTime, default=datetime.utcnow(), nullable=False)
updated_at = Column(DateTime, default=None)
description = Column(Text,default= None)
user_reg = relationship('User',back_populates='roles')
def get_user(user_id):
print("get_user")
user = Session.query(User).filter_by(org_user_id=user_id).first()
print("Active Session", {'id': user.org_user_id, 'email': user.email} )
# print(con , "con of poooolsss start")
# cursor.close()
Session.close()
# print(con.close() , "con of poooolsss")
user = Session.query(User).filter_by(org_user_id=user_id).first()
print("After closing session", {'id': user.org_user_id, 'email':
user.email}
)
--
SQLAlchemy -
The Python SQL Toolkit and Object Relational Mapper
http://www.sqlalchemy.org/
To post example code, please provide an MCVE: Minimal, Complete, and Verifiable
Example. See http://stackoverflow.com/help/mcve for a full description.
---
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To view this discussion on the web visit
https://groups.google.com/d/msgid/sqlalchemy/1312c213-13b8-48d0-9c0c-419edff3da0cn%40googlegroups.com.
