<p align="center">
<a href="https://logyca.com/"><img src="https://logyca.com/sites/default/files/logyca.png" alt="Logyca"></a>
</p>
<p align="center">
<em>LOGYCA public libraries</em>
</p>
<p align="center">
<a href="https://pypi.org/project/logyca" target="_blank">
<img src="https://img.shields.io/pypi/v/logyca?color=orange&label=PyPI%20Package" alt="Package version">
</a>
<a href="(https://www.python.org" target="_blank">
<img src="https://img.shields.io/badge/Python-%5B%3E%3D3.8%2C%3C%3D3.11%5D-orange" alt="Python">
</a>
</p>
---
# About us
* <a href="http://logyca.com" target="_blank">LOGYCA Company</a>
* <a href="https://www.youtube.com/channel/UCzcJtxfScoAtwFbxaLNnEtA" target="_blank">LOGYCA Youtube Channel</a>
* <a href="https://www.linkedin.com/company/logyca" target="_blank"><img src="https://img.shields.io/badge/LinkedIn-0077B5?style=for-the-badge&logo=linkedin&logoColor=white" alt="Linkedin"></a>
* <a href="https://twitter.com/LOGYCA_Org" target="_blank"><img src="https://img.shields.io/badge/Twitter-1DA1F2?style=for-the-badge&logo=twitter&logoColor=white" alt="Twitter"></a>
* <a href="https://www.facebook.com/OrganizacionLOGYCA/" target="_blank"><img src="https://img.shields.io/badge/Facebook-1877F2?style=for-the-badge&logo=facebook&logoColor=white" alt="Facebook"></a>
---
# LOGYCA public libraries: Standard methods to connect to postgres
* **Traversal libraries**: Standard methods to connect to postgres with dependency injection using yield to be used by microservices such as API(s), Workers or scripts.
[Source code](https://github.com/logyca/python-libraries/tree/main/logyca-postgres)
| [Package (PyPI)](https://pypi.org/project/logyca-postgres/)
| [Samples](https://github.com/logyca/python-libraries/tree/main/logyca-postgres/samples)
---
# "pip install" dependency check
The user must select the required libraries and versions for the project that uses this library, which validates that they are pre-installed in order to be installed.
To install the libraries of the logyca postgres package verifying the SQLAlchemy prerequisite without validating connection drivers to postgres, use the following command:
```Python
# Check SQLAlchemy dependency that is installed
pip install logyca_postgres
```
To install the logyca postgres package libraries and validate the postgres asynchronous or synchronous connection driver, use the following command:
```Python
# Check asyncpg driver dependency that is installed
pip install logyca_postgres[async]
# Check psycopg2 driver dependency that is installed
pip install logyca_postgres[sync-psycopg2]
# Check psycopg2-binary driver dependency that is installed
pip install logyca_postgres[sync-psycopg2-binary]
# Check asyncpg+psycopg2-binary driver dependency that is installed
pip install logyca_postgres[async-sync-psycopg2]
```
---
# Semantic Versioning
logyca < MAJOR >.< MINOR >.< PATCH >
* **MAJOR**: version when you make incompatible API changes
* **MINOR**: version when you add functionality in a backwards compatible manner
* **PATCH**: version when you make backwards compatible bug fixes
## Definitions for releasing versions
* https://peps.python.org/pep-0440/
- X.YaN (Alpha release): Identify and fix early-stage bugs. Not suitable for production use.
- X.YbN (Beta release): Stabilize and refine features. Address reported bugs. Prepare for official release.
- X.YrcN (Release candidate): Final version before official release. Assumes all major features are complete and stable. Recommended for testing in non-critical environments.
- X.Y (Final release/Stable/Production): Completed, stable version ready for use in production. Full release for public use.
---
# Example of concepts that use a library with a singleton pattern and connection to multiple engines with yield dependency injection
The library uses a singleton pattern "class SyncConnEngine(metaclass=Singleton):", where the class is allowed to be instantiated only once. You can create another connection to another engine but you must create an inherited class in order to create a new configuration instance.
Example:
class SyncConnEngineX(SyncConnEngine):
def __init__(self, url_connection,server_settings):
super().__init__(url_connection,server_settings)
sync_session_x=SyncConnEngineX(
url_connection=SyncConnEngine.build_url_connection(user=settings.DB_USER_X,password=settings.DB_PASS_X,host=settings.DB_HOST_X,port=settings.DB_PORT_X,database=settings.DB_NAME_X,ssl_enable=settings.DB_SSL_X),
server_settings=SyncConnEngine.server_settings(pool_size=5,max_overflow=1,pool_recycle=10800,application_name=f"{App.Settings.NAME} - SyncConnEngineX")
)
## Asynchronous mode
FastAPI
```python
from fastapi import FastAPI, Depends, HTTPException
from logyca_postgres import AsyncConnEngine, commit_rollback_async, check_connection_async
from sqlalchemy import text as text_to_sql
from sqlalchemy.ext.asyncio import AsyncSession
import os
DB_USER=os.getenv('DB_USER','postgres')
DB_PASS=os.getenv('DB_PASS','xxx')
DB_HOST=os.getenv('DB_HOST','localhost')
DB_PORT=os.getenv('DB_PORT',5432)
DB_NAME=os.getenv('DB_NAME','test')
ssl_enable_like_local_docker_container=False
app = FastAPI()
conn_async_session=AsyncConnEngine(
url_connection=AsyncConnEngine.build_url_connection(user=DB_USER,password=DB_PASS,host=DB_HOST,port=DB_PORT,database=DB_NAME,ssl_enable=ssl_enable_like_local_docker_container),
server_settings=AsyncConnEngine.server_settings(pool_size=5,max_overflow=1,pool_recycle=10800,application_name="MyApp - AsyncConnEngine")
)
'''
The connection pool (pool_size) after the first query will remain open until the application is stopped.
'''
@app.get("/simulated_query_async/")
async def read_item(async_session:AsyncSession = Depends(conn_async_session)):
try:
status, date_time_check_conn = await check_connection_async(async_session)
if(status):
query = text_to_sql("SELECT now();")
result = await async_session.execute(query)
simulated_query = result.scalar_one_or_none()
await commit_rollback_async(async_session)
return {"date_time_check_conn": date_time_check_conn, "simulated_query": simulated_query}
else:
raise HTTPException(status_code=404, detail="async_session connect db error...")
except Exception as e:
raise HTTPException(status_code=404, detail=f"error: {e}")
```
Worker or script
```python
from logyca_postgres import AsyncConnEngine, commit_rollback_async, check_connection_async
from sqlalchemy import text as text_to_sql
from sqlalchemy.ext.asyncio import AsyncSession
import asyncio
import os
DB_USER=os.getenv('DB_USER','postgres')
DB_PASS=os.getenv('DB_PASS','xxx')
DB_HOST=os.getenv('DB_HOST','localhost')
DB_PORT=os.getenv('DB_PORT',5432)
DB_NAME=os.getenv('DB_NAME','test')
ssl_enable_like_local_docker_container=False
conn_async_session=AsyncConnEngine(
url_connection=AsyncConnEngine.build_url_connection(user=DB_USER,password=DB_PASS,host=DB_HOST,port=DB_PORT,database=DB_NAME,ssl_enable=ssl_enable_like_local_docker_container),
server_settings=AsyncConnEngine.server_settings(pool_size=5,max_overflow=1,pool_recycle=10800,application_name="MyApp - AsyncConnEngine")
)
'''
The connection pool (pool_size) after the first query will remain open until the application is stopped or the engine is terminated: close_engine().
'''
async def methods(async_session:AsyncSession):
status, date_time_check_conn = await check_connection_async(async_session)
if(status):
query = text_to_sql("SELECT now();")
result = await async_session.execute(query)
simulated_query = result.scalar_one_or_none()
await commit_rollback_async(async_session)
print(f"date_time_check_conn={date_time_check_conn},simulated_query={simulated_query}")
else:
print("async_session connect db error...")
async def main():
async for async_session in conn_async_session.get_async_session():
await methods(async_session)
await conn_async_session.close_engine()
if __name__ == "__main__":
asyncio.run(main())
```
## synchronous mode
FastAPI
```python
from fastapi import FastAPI, Depends, HTTPException
from logyca_postgres import SyncConnEngine, commit_rollback_sync, check_connection_sync
from sqlalchemy.orm.session import Session
import os
from sqlalchemy import text as text_to_sql
DB_USER=os.getenv('DB_USER','postgres')
DB_PASS=os.getenv('DB_PASS','xxx')
DB_HOST=os.getenv('DB_HOST','localhost')
DB_PORT=os.getenv('DB_PORT',5432)
DB_NAME=os.getenv('DB_NAME','test')
ssl_enable_like_local_docker_container=False
app = FastAPI()
conn_sync_session=SyncConnEngine(
url_connection=SyncConnEngine.build_url_connection(user=DB_USER,password=DB_PASS,host=DB_HOST,port=DB_PORT,database=DB_NAME,ssl_enable=ssl_enable_like_local_docker_container),
server_settings=SyncConnEngine.server_settings(pool_size=5,max_overflow=1,pool_recycle=10800,application_name="MyApp - AsyncConnEngine")
)
'''
The connection pool (pool_size) after the first query will remain open until the application is stopped.
'''
@app.get("/simulated_query_sync/")
def read_item(sync_session:Session = Depends(conn_sync_session)):
try:
status, date_time_check_conn = check_connection_sync(sync_session)
if(status):
query = text_to_sql("SELECT now();")
result = sync_session.execute(query)
simulated_query = result.fetchone()[0]
commit_rollback_sync(sync_session)
return {"date_time_check_conn": date_time_check_conn, "simulated_query": simulated_query}
else:
raise HTTPException(status_code=404, detail="async_session connect db error...")
except Exception as e:
raise HTTPException(status_code=404, detail=f"error: {e}")
```
Worker or script
```python
from logyca_postgres import SyncConnEngine, commit_rollback_sync, check_connection_sync
from sqlalchemy import text as text_to_sql
from sqlalchemy.orm.session import Session
import os
DB_USER=os.getenv('DB_USER','postgres')
DB_PASS=os.getenv('DB_PASS','***')
DB_HOST=os.getenv('DB_HOST','localhost')
DB_PORT=os.getenv('DB_PORT',5432)
DB_NAME=os.getenv('DB_NAME','test')
ssl_enable_like_local_docker_container=False
conn_sync_session=SyncConnEngine(
url_connection=SyncConnEngine.build_url_connection(user=DB_USER,password=DB_PASS,host=DB_HOST,port=DB_PORT,database=DB_NAME,ssl_enable=ssl_enable_like_local_docker_container),
server_settings=SyncConnEngine.server_settings(pool_size=5,max_overflow=1,pool_recycle=10800,application_name="MyApp - SyncConnEngine")
)
'''
The connection pool (pool_size) after the first query will remain open until the application is stopped or the engine is terminated: close_engine().
'''
def methods(sync_session: Session):
status, date_time_check_conn = check_connection_sync(sync_session)
if(status):
query = text_to_sql("SELECT now();")
result = sync_session.execute(query)
simulated_query = result.fetchone()[0]
commit_rollback_sync(sync_session)
print(f"date_time_check_conn={date_time_check_conn},simulated_query={simulated_query}")
else:
print("sync_session connect db error...")
def main():
for sync_session in conn_sync_session.get_sync_session():
methods(sync_session)
conn_sync_session.close_engine()
if __name__ == "__main__":
main()
```
---
# Changelog
All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
## Types of changes
- Added for new features.
- Changed for changes in existing functionality.
- Deprecated for soon-to-be removed features.
- Removed for now removed features.
- Fixed for any bug fixes.
- Security in case of vulnerabilities.
## [0.0.1rc1] - 2024-04-22
### Added
- First tests using pypi.org in develop environment.
- New functionality of asynchronous and synchronous connections to postgresql databases.
- Functionalities can be used in fastapi or workers like Azure Functions.
- Examples of use are added to the documentation of the functions in docstring
- In the samples folder of this library, there are complete working examples of using the code.
## [0.1.0] - 2024-05-21
### Added
- Completion of testing and launch into production.
## [0.1.2] - 2024-05-23
### Added
- Documentation integrated with github
## [0.1.3] - 2024-05-30
### Fixed
- Static messages are removed, since they do not cover errors globally.
Raw data
{
"_id": null,
"home_page": "https://github.com/logyca/python-libraries/tree/main/logyca-postgres",
"name": "logyca-postgres",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": null,
"keywords": "postgres, driver database",
"author": "Jaime Andres Cardona Carrillo",
"author_email": "jacardona@outlook.com",
"download_url": "https://files.pythonhosted.org/packages/90/a3/d58a7cfc07b80c588ce7a2644fdcf3b33ad224af37a933b635ea7961c745/logyca_postgres-0.1.3.tar.gz",
"platform": null,
"description": "<p align=\"center\">\r\n <a href=\"https://logyca.com/\"><img src=\"https://logyca.com/sites/default/files/logyca.png\" alt=\"Logyca\"></a>\r\n</p>\r\n<p align=\"center\">\r\n <em>LOGYCA public libraries</em>\r\n</p>\r\n\r\n<p align=\"center\">\r\n<a href=\"https://pypi.org/project/logyca\" target=\"_blank\">\r\n <img src=\"https://img.shields.io/pypi/v/logyca?color=orange&label=PyPI%20Package\" alt=\"Package version\">\r\n</a>\r\n<a href=\"(https://www.python.org\" target=\"_blank\">\r\n <img src=\"https://img.shields.io/badge/Python-%5B%3E%3D3.8%2C%3C%3D3.11%5D-orange\" alt=\"Python\">\r\n</a>\r\n</p>\r\n\r\n\r\n---\r\n\r\n# About us\r\n\r\n* <a href=\"http://logyca.com\" target=\"_blank\">LOGYCA Company</a>\r\n* <a href=\"https://www.youtube.com/channel/UCzcJtxfScoAtwFbxaLNnEtA\" target=\"_blank\">LOGYCA Youtube Channel</a>\r\n* <a href=\"https://www.linkedin.com/company/logyca\" target=\"_blank\"><img src=\"https://img.shields.io/badge/LinkedIn-0077B5?style=for-the-badge&logo=linkedin&logoColor=white\" alt=\"Linkedin\"></a>\r\n* <a href=\"https://twitter.com/LOGYCA_Org\" target=\"_blank\"><img src=\"https://img.shields.io/badge/Twitter-1DA1F2?style=for-the-badge&logo=twitter&logoColor=white\" alt=\"Twitter\"></a>\r\n* <a href=\"https://www.facebook.com/OrganizacionLOGYCA/\" target=\"_blank\"><img src=\"https://img.shields.io/badge/Facebook-1877F2?style=for-the-badge&logo=facebook&logoColor=white\" alt=\"Facebook\"></a>\r\n\r\n---\r\n\r\n# LOGYCA public libraries: Standard methods to connect to postgres\r\n\r\n* **Traversal libraries**: Standard methods to connect to postgres with dependency injection using yield to be used by microservices such as API(s), Workers or scripts.\r\n\r\n[Source code](https://github.com/logyca/python-libraries/tree/main/logyca-postgres)\r\n| [Package (PyPI)](https://pypi.org/project/logyca-postgres/)\r\n| [Samples](https://github.com/logyca/python-libraries/tree/main/logyca-postgres/samples)\r\n\r\n---\r\n\r\n# \"pip install\" dependency check\r\nThe user must select the required libraries and versions for the project that uses this library, which validates that they are pre-installed in order to be installed.\r\n\r\nTo install the libraries of the logyca postgres package verifying the SQLAlchemy prerequisite without validating connection drivers to postgres, use the following command:\r\n\r\n```Python\r\n# Check SQLAlchemy dependency that is installed\r\npip install logyca_postgres\r\n```\r\n\r\nTo install the logyca postgres package libraries and validate the postgres asynchronous or synchronous connection driver, use the following command:\r\n\r\n```Python\r\n# Check asyncpg driver dependency that is installed\r\npip install logyca_postgres[async]\r\n# Check psycopg2 driver dependency that is installed\r\npip install logyca_postgres[sync-psycopg2]\r\n# Check psycopg2-binary driver dependency that is installed\r\npip install logyca_postgres[sync-psycopg2-binary]\r\n# Check asyncpg+psycopg2-binary driver dependency that is installed\r\npip install logyca_postgres[async-sync-psycopg2]\r\n```\r\n---\r\n\r\n# Semantic Versioning\r\n\r\nlogyca < MAJOR >.< MINOR >.< PATCH >\r\n\r\n* **MAJOR**: version when you make incompatible API changes\r\n* **MINOR**: version when you add functionality in a backwards compatible manner\r\n* **PATCH**: version when you make backwards compatible bug fixes\r\n\r\n## Definitions for releasing versions\r\n* https://peps.python.org/pep-0440/\r\n\r\n - X.YaN (Alpha release): Identify and fix early-stage bugs. Not suitable for production use.\r\n - X.YbN (Beta release): Stabilize and refine features. Address reported bugs. Prepare for official release.\r\n - X.YrcN (Release candidate): Final version before official release. Assumes all major features are complete and stable. Recommended for testing in non-critical environments.\r\n - X.Y (Final release/Stable/Production): Completed, stable version ready for use in production. Full release for public use.\r\n\r\n---\r\n\r\n# Example of concepts that use a library with a singleton pattern and connection to multiple engines with yield dependency injection\r\n\r\nThe library uses a singleton pattern \"class SyncConnEngine(metaclass=Singleton):\", where the class is allowed to be instantiated only once. You can create another connection to another engine but you must create an inherited class in order to create a new configuration instance.\r\n\r\nExample:\r\nclass SyncConnEngineX(SyncConnEngine):\r\n def __init__(self, url_connection,server_settings):\r\n super().__init__(url_connection,server_settings)\r\nsync_session_x=SyncConnEngineX(\r\n url_connection=SyncConnEngine.build_url_connection(user=settings.DB_USER_X,password=settings.DB_PASS_X,host=settings.DB_HOST_X,port=settings.DB_PORT_X,database=settings.DB_NAME_X,ssl_enable=settings.DB_SSL_X),\r\n server_settings=SyncConnEngine.server_settings(pool_size=5,max_overflow=1,pool_recycle=10800,application_name=f\"{App.Settings.NAME} - SyncConnEngineX\")\r\n )\r\n\r\n## Asynchronous mode\r\nFastAPI\r\n```python\r\nfrom fastapi import FastAPI, Depends, HTTPException\r\nfrom logyca_postgres import AsyncConnEngine, commit_rollback_async, check_connection_async\r\nfrom sqlalchemy import text as text_to_sql\r\nfrom sqlalchemy.ext.asyncio import AsyncSession\r\nimport os\r\n\r\nDB_USER=os.getenv('DB_USER','postgres')\r\nDB_PASS=os.getenv('DB_PASS','xxx')\r\nDB_HOST=os.getenv('DB_HOST','localhost')\r\nDB_PORT=os.getenv('DB_PORT',5432)\r\nDB_NAME=os.getenv('DB_NAME','test')\r\nssl_enable_like_local_docker_container=False\r\n\r\napp = FastAPI()\r\n\r\nconn_async_session=AsyncConnEngine(\r\n url_connection=AsyncConnEngine.build_url_connection(user=DB_USER,password=DB_PASS,host=DB_HOST,port=DB_PORT,database=DB_NAME,ssl_enable=ssl_enable_like_local_docker_container),\r\n server_settings=AsyncConnEngine.server_settings(pool_size=5,max_overflow=1,pool_recycle=10800,application_name=\"MyApp - AsyncConnEngine\")\r\n )\r\n\r\n'''\r\nThe connection pool (pool_size) after the first query will remain open until the application is stopped.\r\n'''\r\n\r\n@app.get(\"/simulated_query_async/\")\r\nasync def read_item(async_session:AsyncSession = Depends(conn_async_session)):\r\n try:\r\n status, date_time_check_conn = await check_connection_async(async_session)\r\n if(status):\r\n query = text_to_sql(\"SELECT now();\")\r\n result = await async_session.execute(query)\r\n simulated_query = result.scalar_one_or_none()\r\n await commit_rollback_async(async_session)\r\n return {\"date_time_check_conn\": date_time_check_conn, \"simulated_query\": simulated_query}\r\n else:\r\n raise HTTPException(status_code=404, detail=\"async_session connect db error...\")\r\n except Exception as e:\r\n raise HTTPException(status_code=404, detail=f\"error: {e}\")\r\n```\r\nWorker or script\r\n```python\r\nfrom logyca_postgres import AsyncConnEngine, commit_rollback_async, check_connection_async\r\nfrom sqlalchemy import text as text_to_sql\r\nfrom sqlalchemy.ext.asyncio import AsyncSession\r\nimport asyncio\r\nimport os\r\n\r\nDB_USER=os.getenv('DB_USER','postgres')\r\nDB_PASS=os.getenv('DB_PASS','xxx')\r\nDB_HOST=os.getenv('DB_HOST','localhost')\r\nDB_PORT=os.getenv('DB_PORT',5432)\r\nDB_NAME=os.getenv('DB_NAME','test')\r\nssl_enable_like_local_docker_container=False\r\n\r\nconn_async_session=AsyncConnEngine(\r\n url_connection=AsyncConnEngine.build_url_connection(user=DB_USER,password=DB_PASS,host=DB_HOST,port=DB_PORT,database=DB_NAME,ssl_enable=ssl_enable_like_local_docker_container),\r\n server_settings=AsyncConnEngine.server_settings(pool_size=5,max_overflow=1,pool_recycle=10800,application_name=\"MyApp - AsyncConnEngine\")\r\n )\r\n\r\n'''\r\nThe connection pool (pool_size) after the first query will remain open until the application is stopped or the engine is terminated: close_engine().\r\n'''\r\n\r\nasync def methods(async_session:AsyncSession):\r\n status, date_time_check_conn = await check_connection_async(async_session)\r\n if(status):\r\n query = text_to_sql(\"SELECT now();\")\r\n result = await async_session.execute(query)\r\n simulated_query = result.scalar_one_or_none()\r\n await commit_rollback_async(async_session)\r\n print(f\"date_time_check_conn={date_time_check_conn},simulated_query={simulated_query}\")\r\n else:\r\n print(\"async_session connect db error...\")\r\nasync def main():\r\n async for async_session in conn_async_session.get_async_session():\r\n await methods(async_session)\r\n await conn_async_session.close_engine()\r\n\r\nif __name__ == \"__main__\":\r\n asyncio.run(main())\r\n```\r\n\r\n## synchronous mode\r\nFastAPI\r\n```python\r\nfrom fastapi import FastAPI, Depends, HTTPException\r\nfrom logyca_postgres import SyncConnEngine, commit_rollback_sync, check_connection_sync\r\nfrom sqlalchemy.orm.session import Session\r\nimport os\r\nfrom sqlalchemy import text as text_to_sql\r\n\r\nDB_USER=os.getenv('DB_USER','postgres')\r\nDB_PASS=os.getenv('DB_PASS','xxx')\r\nDB_HOST=os.getenv('DB_HOST','localhost')\r\nDB_PORT=os.getenv('DB_PORT',5432)\r\nDB_NAME=os.getenv('DB_NAME','test')\r\nssl_enable_like_local_docker_container=False\r\n\r\napp = FastAPI()\r\n\r\nconn_sync_session=SyncConnEngine(\r\n url_connection=SyncConnEngine.build_url_connection(user=DB_USER,password=DB_PASS,host=DB_HOST,port=DB_PORT,database=DB_NAME,ssl_enable=ssl_enable_like_local_docker_container),\r\n server_settings=SyncConnEngine.server_settings(pool_size=5,max_overflow=1,pool_recycle=10800,application_name=\"MyApp - AsyncConnEngine\")\r\n )\r\n\r\n'''\r\nThe connection pool (pool_size) after the first query will remain open until the application is stopped.\r\n'''\r\n\r\n@app.get(\"/simulated_query_sync/\")\r\ndef read_item(sync_session:Session = Depends(conn_sync_session)):\r\n try:\r\n status, date_time_check_conn = check_connection_sync(sync_session)\r\n if(status):\r\n query = text_to_sql(\"SELECT now();\")\r\n result = sync_session.execute(query)\r\n simulated_query = result.fetchone()[0]\r\n commit_rollback_sync(sync_session)\r\n return {\"date_time_check_conn\": date_time_check_conn, \"simulated_query\": simulated_query}\r\n else:\r\n raise HTTPException(status_code=404, detail=\"async_session connect db error...\")\r\n except Exception as e:\r\n raise HTTPException(status_code=404, detail=f\"error: {e}\")\r\n```\r\nWorker or script\r\n```python\r\nfrom logyca_postgres import SyncConnEngine, commit_rollback_sync, check_connection_sync\r\nfrom sqlalchemy import text as text_to_sql\r\nfrom sqlalchemy.orm.session import Session\r\nimport os\r\n\r\nDB_USER=os.getenv('DB_USER','postgres')\r\nDB_PASS=os.getenv('DB_PASS','***')\r\nDB_HOST=os.getenv('DB_HOST','localhost')\r\nDB_PORT=os.getenv('DB_PORT',5432)\r\nDB_NAME=os.getenv('DB_NAME','test')\r\nssl_enable_like_local_docker_container=False\r\n\r\nconn_sync_session=SyncConnEngine(\r\n url_connection=SyncConnEngine.build_url_connection(user=DB_USER,password=DB_PASS,host=DB_HOST,port=DB_PORT,database=DB_NAME,ssl_enable=ssl_enable_like_local_docker_container),\r\n server_settings=SyncConnEngine.server_settings(pool_size=5,max_overflow=1,pool_recycle=10800,application_name=\"MyApp - SyncConnEngine\")\r\n )\r\n\r\n'''\r\nThe connection pool (pool_size) after the first query will remain open until the application is stopped or the engine is terminated: close_engine().\r\n'''\r\n\r\ndef methods(sync_session: Session):\r\n status, date_time_check_conn = check_connection_sync(sync_session)\r\n if(status):\r\n query = text_to_sql(\"SELECT now();\")\r\n result = sync_session.execute(query)\r\n simulated_query = result.fetchone()[0]\r\n commit_rollback_sync(sync_session)\r\n print(f\"date_time_check_conn={date_time_check_conn},simulated_query={simulated_query}\")\r\n else:\r\n print(\"sync_session connect db error...\")\r\ndef main():\r\n for sync_session in conn_sync_session.get_sync_session():\r\n methods(sync_session)\r\n conn_sync_session.close_engine() \r\n\r\n\r\nif __name__ == \"__main__\":\r\n main()\r\n```\r\n\r\n---\r\n\r\n# Changelog\r\n\r\nAll notable changes to this project will be documented in this file.\r\n\r\nThe format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),\r\nand this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).\r\n\r\n## Types of changes\r\n\r\n- Added for new features.\r\n- Changed for changes in existing functionality.\r\n- Deprecated for soon-to-be removed features.\r\n- Removed for now removed features.\r\n- Fixed for any bug fixes.\r\n- Security in case of vulnerabilities.\r\n\r\n## [0.0.1rc1] - 2024-04-22\r\n### Added\r\n- First tests using pypi.org in develop environment.\r\n- New functionality of asynchronous and synchronous connections to postgresql databases.\r\n- Functionalities can be used in fastapi or workers like Azure Functions.\r\n- Examples of use are added to the documentation of the functions in docstring\r\n- In the samples folder of this library, there are complete working examples of using the code.\r\n\r\n## [0.1.0] - 2024-05-21\r\n### Added\r\n- Completion of testing and launch into production.\r\n\r\n## [0.1.2] - 2024-05-23\r\n### Added\r\n- Documentation integrated with github\r\n\r\n## [0.1.3] - 2024-05-30\r\n### Fixed\r\n- Static messages are removed, since they do not cover errors globally.\r\n",
"bugtrack_url": null,
"license": "MIT License",
"summary": "An integration package created by the company LOGYCA that connects Postgres and is used to standardize connections and dependency injection in synchronous or asynchronous mode. Tested in fastapi and in console/worker scripts.",
"version": "0.1.3",
"project_urls": {
"Homepage": "https://github.com/logyca/python-libraries/tree/main/logyca-postgres"
},
"split_keywords": [
"postgres",
" driver database"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "28f55a759b60255514285cf954e53d31501af34829d24b1eed9376343a52d75d",
"md5": "b1191baf26ddef4783717d4d85123644",
"sha256": "1461d97464acf2ea8d3f2338f38b9718268cc45106cbfa5f1cb7bac4c5ed4438"
},
"downloads": -1,
"filename": "logyca_postgres-0.1.3-py3-none-any.whl",
"has_sig": false,
"md5_digest": "b1191baf26ddef4783717d4d85123644",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 12963,
"upload_time": "2024-05-30T20:22:45",
"upload_time_iso_8601": "2024-05-30T20:22:45.577766Z",
"url": "https://files.pythonhosted.org/packages/28/f5/5a759b60255514285cf954e53d31501af34829d24b1eed9376343a52d75d/logyca_postgres-0.1.3-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "90a3d58a7cfc07b80c588ce7a2644fdcf3b33ad224af37a933b635ea7961c745",
"md5": "8615290e6b9b4c4b2299accda2fdcb07",
"sha256": "ab403a317a290fbcc40d80f71cb37c14a0796fd9d44f99ac5f626c35a93095f8"
},
"downloads": -1,
"filename": "logyca_postgres-0.1.3.tar.gz",
"has_sig": false,
"md5_digest": "8615290e6b9b4c4b2299accda2fdcb07",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 11945,
"upload_time": "2024-05-30T20:22:47",
"upload_time_iso_8601": "2024-05-30T20:22:47.161309Z",
"url": "https://files.pythonhosted.org/packages/90/a3/d58a7cfc07b80c588ce7a2644fdcf3b33ad224af37a933b635ea7961c745/logyca_postgres-0.1.3.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-05-30 20:22:47",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "logyca",
"github_project": "python-libraries",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "logyca-postgres"
}
JSON
