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 sqlalchemy+unsubscr...@googlegroups.com. To view this discussion on the web visit https://groups.google.com/d/msgid/sqlalchemy/1312c213-13b8-48d0-9c0c-419edff3da0cn%40googlegroups.com.