Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- from psycopg2 import pool
- class PostgresConnectionPool( pool.SimpleConnectionPool ):
- def __init__( self, init_con_number, max_con_number, host, port, database, user, password ):
- # initialize the Postgresql connection pool object
- super().__init__( minconn=init_con_number,
- maxconn=max_con_number,
- host=host,
- port=port,
- database=database,
- user=user,
- password=password )
- def get_con_from_pool( self ):
- return self.getconn()
- def put_con_back_to_pool( self, con):
- self.putconn(con)
- # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
- # Example how pool will look:
- # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
- def print_pool_state( name, pool: PostgresConnectionPool):
- print(f"\n===============================\nPool {name} state:\n===============================\n")
- for p, v in pool.__dict__.items():
- if p == '_pool':
- print('_pool connections at instantiation')
- [print(f" {obj}") for obj in v]
- continue
- print(p, v)
- my_pool_2 = PostgresConnectionPool('2', '5','192.168.144.13' ,5432,'softuni_management_db', 'test_name', 'test_pass')
- print_pool_state( 'my_pool_2', my_pool_2)
- # Get free connections from the pool:
- my_connection_1 = my_pool_2.get_con_from_pool()
- my_connection_2 = my_pool_2.get_con_from_pool()
- my_connection_3 = my_pool_2.get_con_from_pool()
- my_connection_4 = my_pool_2.get_con_from_pool()
- my_connection_5 = my_pool_2.get_con_from_pool()
- print_pool_state( 'my_pool_2', my_pool_2)
- try:
- unsuccessful_con = my_pool_2.get_con_from_pool()
- except Exception as e:
- print(f"\n===============================\n"
- f"Pool {my_pool_2.__class__.__name__} state:\n"
- f"===============================\n{e}")
- my_pool_2.put_con_back_to_pool(my_connection_1)
- print_pool_state( 'my_pool_2', my_pool_2)
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement
