# repository_sqlalchemy
repository_sqlalchemy is a small library that simplifies SQLAlchemy usage with automatic transaction and session management. It provides a base repository pattern implementation, similar to JPA in the Java world.
## Features
- Base repository pattern for common database operations
- Automatic session management using context variables
- Transaction management with a convenient decorator
- Support for nested transactions and savepoints
- No need for manual session commits or rollbacks
## Installation
You can install repository_sqlalchemy using pip:
```bash
pip install repository_sqlalchemy
```
## Usage
### Environment Setup
Before using the library, set up the following environment variables:
```bash
export DB_TYPE=postgresql # or mysql, sqlite
export DB_USER=your_username
export DB_PASSWORD=your_password
export DB_HOST=your_host
export DB_PORT=your_port
export DB_NAME=your_database_name
```
### Example Usage
Here's a comprehensive example demonstrating how to use the repository_sqlalchemy library:
```python
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from repository_sqlalchemy import BaseRepository, Base, transaction
from typing import List, Dict, Any
class UserModel(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
username = Column(String(50), unique=True, nullable=False)
email = Column(String(120), unique=True, nullable=False)
class PostModel(Base):
__tablename__ = 'posts'
id = Column(Integer, primary_key=True)
title = Column(String(100), nullable=False)
content = Column(String(1000), nullable=False)
user_id = Column(Integer, ForeignKey('users.id'), nullable=False)
class UserRepository(BaseRepository[UserModel]):
def find_by_username(self, username: str) -> UserModel:
return self.session.query(self.model).filter_by(username=username).first()
class PostRepository(BaseRepository[PostModel]):
def find_by_user_id(self, user_id: int) -> List[PostModel]:
return self.session.query(self.model).filter_by(user_id=user_id).all()
@transaction()
def create_user_with_posts(username: str, email: str, posts: List[Dict[str, str]]) -> UserModel:
user_repo = UserRepository()
post_repo = PostRepository()
# Create user
user = user_repo.create(UserModel(username=username, email=email))
# Create posts for the user
for post_data in posts:
post = PostModel(title=post_data['title'], content=post_data['content'], user_id=user.id)
post_repo.create(post)
return user
@transaction()
def update_user_and_posts(username: str, user_data: Dict[str, Any], post_updates: List[Dict[str, Any]]) -> UserModel:
user_repo = UserRepository()
post_repo = PostRepository()
user = user_repo.find_by_username(username)
if not user:
raise ValueError(f"User {username} not found")
# Update user
user_repo.update(user, user_data)
# Update posts
posts = post_repo.find_by_user_id(user.id)
for post, post_data in zip(posts, post_updates):
post_repo.update(post, post_data)
return user
# Usage
new_user = create_user_with_posts(
"john_doe",
"john@example.com",
[
{"title": "First Post", "content": "Hello, world!"},
{"title": "Second Post", "content": "This is my second post."}
]
)
print(f"Created user: {new_user.username} with 2 posts")
updated_user = update_user_and_posts(
"john_doe",
{"email": "john.doe@newdomain.com"},
[
{"title": "Updated First Post"},
{"content": "Updated content for second post."}
]
)
print(f"Updated user: {updated_user.username}, {updated_user.email}")
```
This example demonstrates:
1. Defining SQLAlchemy models (`UserModel` and `PostModel`).
2. Creating custom repositories (`UserRepository` and `PostRepository`) with additional methods.
3. Using the `@transaction()` decorator for automatic transaction management across multiple operations.
4. Performing create and update operations on multiple entities within a single transaction.
5. Automatic session handling within repository methods, with no need for manual commits or rollbacks.
## Contributing
Contributions are welcome! Please feel free to submit a Pull Request.
## License
This project is licensed under the MIT License.
Raw data
{
"_id": null,
"home_page": "https://github.com/ryan-zheng-teki/repository_sqlalchemy",
"name": "repository-sqlalchemy",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": null,
"keywords": null,
"author": "Ryan Zheng",
"author_email": "ryan.zheng.work@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/27/0a/7da2db2909868dc6b355eea22b27f77e1912b9ea664c69071d17a771a9ae/repository_sqlalchemy-0.1.7.tar.gz",
"platform": null,
"description": "# repository_sqlalchemy\n\nrepository_sqlalchemy is a small library that simplifies SQLAlchemy usage with automatic transaction and session management. It provides a base repository pattern implementation, similar to JPA in the Java world.\n\n## Features\n\n- Base repository pattern for common database operations\n- Automatic session management using context variables\n- Transaction management with a convenient decorator\n- Support for nested transactions and savepoints\n- No need for manual session commits or rollbacks\n\n## Installation\n\nYou can install repository_sqlalchemy using pip:\n\n```bash\npip install repository_sqlalchemy\n```\n\n## Usage\n\n### Environment Setup\n\nBefore using the library, set up the following environment variables:\n\n```bash\nexport DB_TYPE=postgresql # or mysql, sqlite\nexport DB_USER=your_username\nexport DB_PASSWORD=your_password\nexport DB_HOST=your_host\nexport DB_PORT=your_port\nexport DB_NAME=your_database_name\n```\n\n### Example Usage\n\nHere's a comprehensive example demonstrating how to use the repository_sqlalchemy library:\n\n```python\nfrom sqlalchemy import Column, Integer, String, ForeignKey\nfrom sqlalchemy.ext.declarative import declarative_base\nfrom repository_sqlalchemy import BaseRepository, Base, transaction\nfrom typing import List, Dict, Any\n\n\nclass UserModel(Base):\n __tablename__ = 'users'\n id = Column(Integer, primary_key=True)\n username = Column(String(50), unique=True, nullable=False)\n email = Column(String(120), unique=True, nullable=False)\n\nclass PostModel(Base):\n __tablename__ = 'posts'\n id = Column(Integer, primary_key=True)\n title = Column(String(100), nullable=False)\n content = Column(String(1000), nullable=False)\n user_id = Column(Integer, ForeignKey('users.id'), nullable=False)\n\nclass UserRepository(BaseRepository[UserModel]):\n def find_by_username(self, username: str) -> UserModel:\n return self.session.query(self.model).filter_by(username=username).first()\n\nclass PostRepository(BaseRepository[PostModel]):\n def find_by_user_id(self, user_id: int) -> List[PostModel]:\n return self.session.query(self.model).filter_by(user_id=user_id).all()\n\n@transaction()\ndef create_user_with_posts(username: str, email: str, posts: List[Dict[str, str]]) -> UserModel:\n user_repo = UserRepository()\n post_repo = PostRepository()\n\n # Create user\n user = user_repo.create(UserModel(username=username, email=email))\n\n # Create posts for the user\n for post_data in posts:\n post = PostModel(title=post_data['title'], content=post_data['content'], user_id=user.id)\n post_repo.create(post)\n\n return user\n\n@transaction()\ndef update_user_and_posts(username: str, user_data: Dict[str, Any], post_updates: List[Dict[str, Any]]) -> UserModel:\n user_repo = UserRepository()\n post_repo = PostRepository()\n\n user = user_repo.find_by_username(username)\n if not user:\n raise ValueError(f\"User {username} not found\")\n\n # Update user\n user_repo.update(user, user_data)\n\n # Update posts\n posts = post_repo.find_by_user_id(user.id)\n for post, post_data in zip(posts, post_updates):\n post_repo.update(post, post_data)\n\n return user\n\n# Usage\nnew_user = create_user_with_posts(\n \"john_doe\",\n \"john@example.com\",\n [\n {\"title\": \"First Post\", \"content\": \"Hello, world!\"},\n {\"title\": \"Second Post\", \"content\": \"This is my second post.\"}\n ]\n)\nprint(f\"Created user: {new_user.username} with 2 posts\")\n\nupdated_user = update_user_and_posts(\n \"john_doe\",\n {\"email\": \"john.doe@newdomain.com\"},\n [\n {\"title\": \"Updated First Post\"},\n {\"content\": \"Updated content for second post.\"}\n ]\n)\nprint(f\"Updated user: {updated_user.username}, {updated_user.email}\")\n```\n\nThis example demonstrates:\n\n1. Defining SQLAlchemy models (`UserModel` and `PostModel`).\n2. Creating custom repositories (`UserRepository` and `PostRepository`) with additional methods.\n3. Using the `@transaction()` decorator for automatic transaction management across multiple operations.\n4. Performing create and update operations on multiple entities within a single transaction.\n5. Automatic session handling within repository methods, with no need for manual commits or rollbacks.\n\n## Contributing\n\nContributions are welcome! Please feel free to submit a Pull Request.\n\n## License\n\nThis project is licensed under the MIT License.\n",
"bugtrack_url": null,
"license": null,
"summary": "A small library to simplify SQLAlchemy usage with FastAPI",
"version": "0.1.7",
"project_urls": {
"Homepage": "https://github.com/ryan-zheng-teki/repository_sqlalchemy"
},
"split_keywords": [],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "5b19ecf9b71b7894c8812e0d50e85cf3df8f46924934de034e2a5d9bf0f8846d",
"md5": "3b89e9fe28aaac9a893fbe762b83eb89",
"sha256": "4c53b8a92de5d8718efb65890cf55038b8d28ff85aefa91d313a8a522016bc6a"
},
"downloads": -1,
"filename": "repository_sqlalchemy-0.1.7-py3-none-any.whl",
"has_sig": false,
"md5_digest": "3b89e9fe28aaac9a893fbe762b83eb89",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 10947,
"upload_time": "2025-07-10T19:31:56",
"upload_time_iso_8601": "2025-07-10T19:31:56.952004Z",
"url": "https://files.pythonhosted.org/packages/5b/19/ecf9b71b7894c8812e0d50e85cf3df8f46924934de034e2a5d9bf0f8846d/repository_sqlalchemy-0.1.7-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "270a7da2db2909868dc6b355eea22b27f77e1912b9ea664c69071d17a771a9ae",
"md5": "f272eb9a3a93580e2745a9510e60ff20",
"sha256": "7ab55f9e6b6353ba6c3e36e4cb47a51fbffd7b65d75f28adf44dea60e62a1ba1"
},
"downloads": -1,
"filename": "repository_sqlalchemy-0.1.7.tar.gz",
"has_sig": false,
"md5_digest": "f272eb9a3a93580e2745a9510e60ff20",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 10089,
"upload_time": "2025-07-10T19:31:57",
"upload_time_iso_8601": "2025-07-10T19:31:57.886399Z",
"url": "https://files.pythonhosted.org/packages/27/0a/7da2db2909868dc6b355eea22b27f77e1912b9ea664c69071d17a771a9ae/repository_sqlalchemy-0.1.7.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-10 19:31:57",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "ryan-zheng-teki",
"github_project": "repository_sqlalchemy",
"github_not_found": true,
"lcname": "repository-sqlalchemy"
}
JSON
