The session-repository library is a Python library that is designed to use the session/repository pattern to interact with databases in Python projects. It provides a more flexible notation for running SQL queries and is built on top of SQLAlchemy, a popular Python SQL toolkit. With session-repository, users can write SQL queries using a new, more intuitive syntax, simplifying the process of working with SQL databases in Python and making it easier to write and maintain complex queries.
## Installing Session-Repository
To install session-repository, if you already have Python, you can install with:
```
pip install session-repository
```
## How to import Session-Repository
To access session-repository and its functions import it in your Python code like this:
```
from session_repository import SessionRepository, SessionService, with_session, Operators, LoadingTechnique
from session_repository.utils import RelationshipOption
```
## Reading the example code
To create a repository, you just have to inherit your class from SessionRepository.
```
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class City(Base):
__tablename__ = "CITY"
id = Column(
"ID",
Integer,
primary_key=True,
index=True,
)
name = Column(
"NAME",
String,
index=True,
)
state = Column(
"STATE",
String,
index=True,
)
addresses = relationship(
"Address",
back_populates="city",
)
class Address(Base):
__tablename__ = "ADDRESS"
id = Column(
"ID",
Integer,
primary_key=True,
index=True,
)
street = Column(
"STREET",
String,
index=True,
)
zip_code = Column(
"ZIP_CODE",
Integer,
index=True,
)
user_id = Column(
"USER_ID",
Integer,
ForeignKey("USER.ID"),
)
city_id = Column(
"CITY_ID",
Integer,
ForeignKey("CITY.ID"),
)
user = relationship(
"User",
back_populates="addresses",
)
city = relationship(
"City",
back_populates="addresses",
)
class User(Base):
__tablename__ = "USER"
id = Column(
"ID",
Integer,
primary_key=True,
index=True,
)
email = Column(
"EMAIL",
String,
unique=True,
index=True,
)
hashed_password = Column(
"HASHED_PASSWORD",
String,
)
full_name = Column(
"FULL_NAME",
String,
index=True,
)
is_active = Column(
"IS_ACTIVE",
Boolean,
default=True,
)
addresses = relationship(
"Address",
back_populates="user",
)
class UserRepository(SessionRepository):
def __init__(
self,
session_factory: Callable[..., AbstractContextManager[Session]],
) -> None:
super().__init__(session_factory)
@classmethod
def __get_filters(
cls,
ids: Optional[List[int]] = None,
):
return {
User.id: {
Operators.IN: ids,
}
}
@classmethod
def __get_relationship_options(cls):
return {
User.adresses: RelationshipOption(
lazy=LoadingTechnique.JOINED,
children={
Adress.city: RelationshipOption(
lazy=LoadingTechnique.LAZY
)
},
),
}
def get_all(
self,
ids: Optional[List[int]] = None,
order_by: Optional[List[str]] = None,
direction: Optional[List[str]] = None,
current_session: Optional[Session] = None,
) -> List[User]:
users = self._select_all(
current_session=current_session,
model=User,
optional_filters=self.__get_filters(
ids=ids;
),
relationship_options=self.__get_relationship_options(),
order_by=order_by,
direction=direction,
)
return users
def get_paginate(
self,
page: int,
per_page: int,
ids: Optional[List[int]] = None,
order_by: Optional[List[str]] = None,
direction: Optional[List[str]] = None,
current_session: Optional[Session] = None,
) -> Tuple[List[User], str]:
users, pagination = self._select_paginate(
page=page,
per_page=per_page,
model=User,
optional_filters=self.__get_filters(
ids=ids,
),
relationship_options=self.__get_relationship_options(),
order_by=order_by,
direction=direction,
current_session=current_session,
)
return users, pagination
def get_by_id(
self,
id: int,
current_session: Optional[Session] = None,
) -> Optional[User]:
user = self._select(
current_session=current_session,
model=User,
filters={
User.id: {
Operators.EQUAL: id,
},
},
)
return user
def get_by_email(
self,
email: str,
current_session: Optional[Session] = None,
) -> Optional[User]:
user = self._select(
current_session=current_session,
model=User,
filters={
User.email: {
Operators.IEQUAL: email,
},
},
)
return user
def create(
self,
data: UserCreateSchema,
flush: bool = False,
commit: bool = True,
current_session: Optional[Session] = None,
) -> User:
user = self._add(
data=User(
**{
User.email.key: data.email,
User.hashed_password.key: data.hashed_password,
User.full_name.key: data.full_name,
}
),
flush=flush,
commit=commit,
current_session=current_session,
)
return user
def patch_active(
self,
id: int,
is_active: bool,
flush: bool = False,
commit: bool = True,
current_session: Optional[Session] = None,
) -> User:
user = self._update(
current_session=current_session,
model=User,
values={
User.is_active.key: is_active,
},
filters={
User.id: {
Operators.EQUAL: id,
},
},
flush=flush,
commit=commit,
)
return user
def delete(
self,
id: int,
flush: bool = False,
commit: bool = True,
current_session: Optional[Session] = None,
) -> bool:
is_deleted = self._delete(
model=User,
filters={
User.id: {
Operators.EQUAL: id,
},
},
flush=flush,
commit=commit,
current_session=current_session,
)
return is_deleted
```
To create a service, you just have to inherit your class from SessionService.
```
T = TypeVar("T", bound=UserReadSchema)
class UserService(SessionService[UserRepository]):
def __init__(
self,
repository: UserRepository,
logger: Logger,
) -> None:
super().__init__(
repository=repository,
logger=logger,
)
@with_session()
def get_users(
self,
ids: Optional[List[int]] = None,
order_by: Optional[List[str]] = None,
direction: Optional[List[str]] = None,
schema: Type[T] = UserReadSchema,
current_session: Optional[Session] = None,
) -> List[T]:
users = self._repository.get_all(
ids=ids,
order_by=order_by,
direction=direction,
current_session=current_session,
)
return [schema.model_validate(user) for user in users]
@with_session()
def get_users_paginate(
self,
page: int,
per_page: int,
user_permissions: List[str],
ids: Optional[List[int]] = None,
order_by: Optional[List[str]] = None,
direction: Optional[List[str]] = None,
schema: Type[T] = RecipeInspectionFviReadSchema,
current_session: Optional[Session] = None,
) -> Tuple[List[T], str]:
users, pagination = self._repository.get_paginate(
page=page,
per_page=per_page,
ids=ids,
order_by=order_by,
direction=direction,
current_session=current_session,
)
return [schema.model_validate(user) for user in users], pagination
@with_session()
def get_user_by_id(
self,
id: int,
schema: Type[T] = UserReadSchema,
current_session: Optional[Session] = None,
) -> T:
user = self._repository.get_by_id(
id=id,
current_session=current_session,
)
if user is None:
raise ValueError("User not found")
return schema.model_validate(user)
@with_session()
async def create_user(
self,
data: UserCreateSchema,
commit: bool = True,
schema: Type[T] = UserReadSchema,
current_session: Optional[Session] = None,
) -> T:
user = self._repository.get_by_email(
email=data.email,
current_session=current_session,
)
if user is not None:
self._logger.error(
"Unable to create new user beacuse email alrady used bu another one"
)
raise ValueError(
"User already exists with same email"
)
user = self._repository.create(
data=data,
flush=True,
commit=False,
current_session=current_session,
)
return schema.model_validate(user)
@with_session()
async def delete_user(
self,
id: int,
commit: bool = True,
current_session: Optional[Session] = None,
) -> bool:
current_user = self._repository.get_by_id(
id=id,
current_session=current_session,
)
if current_user is None:
raise ValueError(f"No user with {id=}")
is_deleted = self._repository.delete(
id=id,
flush=True,
commit=False,
current_session=current_session,
)
if commit:
current_session.commit()
self._logger.info(
f"User was successfully deleted"
)
return is_deleted
```
Raw data
{
"_id": null,
"home_page": "https://github.com/Impro02/session-repository",
"name": "session-repository",
"maintainer": "",
"docs_url": null,
"requires_python": "",
"maintainer_email": "",
"keywords": "",
"author": "Maxime MARTIN",
"author_email": "maxime.martin02@hotmail.fr",
"download_url": "https://files.pythonhosted.org/packages/0b/42/1d16b1f40f0c69d19f4cf0b00014f32f5f5d87ee79f11a0fcd3538520068/session-repository-0.4.5.4.tar.gz",
"platform": null,
"description": "The session-repository library is a Python library that is designed to use the session/repository pattern to interact with databases in Python projects. It provides a more flexible notation for running SQL queries and is built on top of SQLAlchemy, a popular Python SQL toolkit. With session-repository, users can write SQL queries using a new, more intuitive syntax, simplifying the process of working with SQL databases in Python and making it easier to write and maintain complex queries.\n\n## Installing Session-Repository\n\nTo install session-repository, if you already have Python, you can install with:\n\n```\npip install session-repository\n```\n\n## How to import Session-Repository\n\nTo access session-repository and its functions import it in your Python code like this:\n\n```\nfrom session_repository import SessionRepository, SessionService, with_session, Operators, LoadingTechnique\nfrom session_repository.utils import RelationshipOption\n```\n\n## Reading the example code\n\nTo create a repository, you just have to inherit your class from SessionRepository.\n\n```\nfrom sqlalchemy import Boolean, Column, ForeignKey, Integer, String\nfrom sqlalchemy.orm import relationship\nfrom sqlalchemy.ext.declarative import declarative_base\n\nBase = declarative_base()\n\n\nclass City(Base):\n __tablename__ = \"CITY\"\n\n id = Column(\n \"ID\",\n Integer,\n primary_key=True,\n index=True,\n )\n name = Column(\n \"NAME\",\n String,\n index=True,\n )\n state = Column(\n \"STATE\",\n String,\n index=True,\n )\n\n addresses = relationship(\n \"Address\",\n back_populates=\"city\",\n )\n\n\nclass Address(Base):\n __tablename__ = \"ADDRESS\"\n\n id = Column(\n \"ID\",\n Integer,\n primary_key=True,\n index=True,\n )\n street = Column(\n \"STREET\",\n String,\n index=True,\n )\n zip_code = Column(\n \"ZIP_CODE\",\n Integer,\n index=True,\n )\n user_id = Column(\n \"USER_ID\",\n Integer,\n ForeignKey(\"USER.ID\"),\n )\n city_id = Column(\n \"CITY_ID\",\n Integer,\n ForeignKey(\"CITY.ID\"),\n )\n\n user = relationship(\n \"User\",\n back_populates=\"addresses\",\n )\n city = relationship(\n \"City\",\n back_populates=\"addresses\",\n )\n\n\nclass User(Base):\n __tablename__ = \"USER\"\n\n id = Column(\n \"ID\",\n Integer,\n primary_key=True,\n index=True,\n )\n email = Column(\n \"EMAIL\",\n String,\n unique=True,\n index=True,\n )\n hashed_password = Column(\n \"HASHED_PASSWORD\",\n String,\n )\n full_name = Column(\n \"FULL_NAME\",\n String,\n index=True,\n )\n is_active = Column(\n \"IS_ACTIVE\",\n Boolean,\n default=True,\n )\n\n addresses = relationship(\n \"Address\",\n back_populates=\"user\",\n )\n\n\nclass UserRepository(SessionRepository):\n def __init__(\n self,\n session_factory: Callable[..., AbstractContextManager[Session]],\n ) -> None:\n super().__init__(session_factory)\n\n @classmethod\n def __get_filters(\n cls,\n ids: Optional[List[int]] = None,\n ):\n return {\n User.id: {\n Operators.IN: ids,\n }\n }\n\n @classmethod\n def __get_relationship_options(cls):\n return {\n User.adresses: RelationshipOption(\n lazy=LoadingTechnique.JOINED,\n children={\n Adress.city: RelationshipOption(\n lazy=LoadingTechnique.LAZY\n )\n },\n ),\n }\n\n def get_all(\n self,\n ids: Optional[List[int]] = None,\n order_by: Optional[List[str]] = None,\n direction: Optional[List[str]] = None,\n current_session: Optional[Session] = None,\n ) -> List[User]:\n users = self._select_all(\n current_session=current_session,\n model=User,\n optional_filters=self.__get_filters(\n ids=ids;\n ),\n relationship_options=self.__get_relationship_options(),\n order_by=order_by,\n direction=direction,\n )\n\n return users\n\n def get_paginate(\n self,\n page: int,\n per_page: int,\n ids: Optional[List[int]] = None,\n order_by: Optional[List[str]] = None,\n direction: Optional[List[str]] = None,\n current_session: Optional[Session] = None,\n ) -> Tuple[List[User], str]:\n users, pagination = self._select_paginate(\n page=page,\n per_page=per_page,\n model=User,\n optional_filters=self.__get_filters(\n ids=ids,\n ),\n relationship_options=self.__get_relationship_options(),\n order_by=order_by,\n direction=direction,\n current_session=current_session,\n )\n\n return users, pagination\n\n def get_by_id(\n self,\n id: int,\n current_session: Optional[Session] = None,\n ) -> Optional[User]:\n user = self._select(\n current_session=current_session,\n model=User,\n filters={\n User.id: {\n Operators.EQUAL: id,\n },\n },\n )\n\n return user\n\n def get_by_email(\n self,\n email: str,\n current_session: Optional[Session] = None,\n ) -> Optional[User]:\n user = self._select(\n current_session=current_session,\n model=User,\n filters={\n User.email: {\n Operators.IEQUAL: email,\n },\n },\n )\n\n return user\n\n def create(\n self,\n data: UserCreateSchema,\n flush: bool = False,\n commit: bool = True,\n current_session: Optional[Session] = None,\n ) -> User:\n user = self._add(\n data=User(\n **{\n User.email.key: data.email,\n User.hashed_password.key: data.hashed_password,\n User.full_name.key: data.full_name,\n }\n ),\n flush=flush,\n commit=commit,\n current_session=current_session,\n )\n\n return user\n\n def patch_active(\n self,\n id: int,\n is_active: bool,\n flush: bool = False,\n commit: bool = True,\n current_session: Optional[Session] = None,\n ) -> User:\n user = self._update(\n current_session=current_session,\n model=User,\n values={\n User.is_active.key: is_active,\n },\n filters={\n User.id: {\n Operators.EQUAL: id,\n },\n },\n flush=flush,\n commit=commit,\n )\n\n return user\n\n def delete(\n self,\n id: int,\n flush: bool = False,\n commit: bool = True,\n current_session: Optional[Session] = None,\n ) -> bool:\n is_deleted = self._delete(\n model=User,\n filters={\n User.id: {\n Operators.EQUAL: id,\n },\n },\n flush=flush,\n commit=commit,\n current_session=current_session,\n )\n\n return is_deleted\n```\n\n\nTo create a service, you just have to inherit your class from SessionService.\n\n```\nT = TypeVar(\"T\", bound=UserReadSchema)\n\nclass UserService(SessionService[UserRepository]):\n\n def __init__(\n self,\n repository: UserRepository,\n logger: Logger,\n ) -> None:\n super().__init__(\n repository=repository,\n logger=logger,\n )\n\n\n @with_session()\n def get_users(\n self,\n ids: Optional[List[int]] = None,\n order_by: Optional[List[str]] = None,\n direction: Optional[List[str]] = None,\n schema: Type[T] = UserReadSchema,\n current_session: Optional[Session] = None,\n ) -> List[T]:\n users = self._repository.get_all(\n ids=ids,\n order_by=order_by,\n direction=direction,\n current_session=current_session,\n )\n\n return [schema.model_validate(user) for user in users]\n\n @with_session()\n def get_users_paginate(\n self,\n page: int,\n per_page: int,\n user_permissions: List[str],\n ids: Optional[List[int]] = None,\n order_by: Optional[List[str]] = None,\n direction: Optional[List[str]] = None,\n schema: Type[T] = RecipeInspectionFviReadSchema,\n current_session: Optional[Session] = None,\n ) -> Tuple[List[T], str]:\n users, pagination = self._repository.get_paginate(\n page=page,\n per_page=per_page,\n ids=ids,\n order_by=order_by,\n direction=direction,\n current_session=current_session,\n )\n\n return [schema.model_validate(user) for user in users], pagination\n\n\n @with_session()\n def get_user_by_id(\n self,\n id: int,\n schema: Type[T] = UserReadSchema,\n current_session: Optional[Session] = None,\n ) -> T:\n user = self._repository.get_by_id(\n id=id,\n current_session=current_session,\n )\n\n if user is None:\n raise ValueError(\"User not found\")\n\n return schema.model_validate(user)\n\n @with_session()\n async def create_user(\n self,\n data: UserCreateSchema,\n commit: bool = True,\n schema: Type[T] = UserReadSchema,\n current_session: Optional[Session] = None,\n ) -> T:\n user = self._repository.get_by_email(\n email=data.email,\n current_session=current_session,\n )\n\n if user is not None:\n self._logger.error(\n \"Unable to create new user beacuse email alrady used bu another one\"\n )\n\n raise ValueError(\n \"User already exists with same email\"\n )\n\n user = self._repository.create(\n data=data,\n flush=True,\n commit=False,\n current_session=current_session,\n )\n\n return schema.model_validate(user)\n\n\n @with_session()\n async def delete_user(\n self,\n id: int,\n commit: bool = True,\n current_session: Optional[Session] = None,\n ) -> bool:\n current_user = self._repository.get_by_id(\n id=id,\n current_session=current_session,\n )\n\n if current_user is None:\n raise ValueError(f\"No user with {id=}\")\n\n is_deleted = self._repository.delete(\n id=id,\n flush=True,\n commit=False,\n current_session=current_session,\n )\n\n if commit:\n current_session.commit()\n\n self._logger.info(\n f\"User was successfully deleted\"\n )\n\n return is_deleted\n```\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "A project to have a base repository class to perform select/insert/update/delete with dynamic syntax",
"version": "0.4.5.4",
"project_urls": {
"Download": "https://github.com/Impro02/session-repository/archive/refs/tags/0.4.5.4.tar.gz",
"Homepage": "https://github.com/Impro02/session-repository"
},
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "f623940ce34f60e08c06dbb631b80dfaf142e96b65f2ee0d1e63f41c75c3aaf9",
"md5": "04fa0e6bd8b6633c250be3c15dd5f7ff",
"sha256": "b5e3e5f6ad76a555a746022eda4b7acfd1afbf577b927fe5746cdf80112b3459"
},
"downloads": -1,
"filename": "session_repository-0.4.5.4-py3-none-any.whl",
"has_sig": false,
"md5_digest": "04fa0e6bd8b6633c250be3c15dd5f7ff",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": null,
"size": 9377,
"upload_time": "2024-01-11T15:15:47",
"upload_time_iso_8601": "2024-01-11T15:15:47.439902Z",
"url": "https://files.pythonhosted.org/packages/f6/23/940ce34f60e08c06dbb631b80dfaf142e96b65f2ee0d1e63f41c75c3aaf9/session_repository-0.4.5.4-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "0b421d16b1f40f0c69d19f4cf0b00014f32f5f5d87ee79f11a0fcd3538520068",
"md5": "4780cd296fd6723efe5097678c27b865",
"sha256": "8cfed470e3672d6ab426345ee0b6a3b84a9602780b8fe746f137ecfc2b442841"
},
"downloads": -1,
"filename": "session-repository-0.4.5.4.tar.gz",
"has_sig": false,
"md5_digest": "4780cd296fd6723efe5097678c27b865",
"packagetype": "sdist",
"python_version": "source",
"requires_python": null,
"size": 9799,
"upload_time": "2024-01-11T15:15:52",
"upload_time_iso_8601": "2024-01-11T15:15:52.131319Z",
"url": "https://files.pythonhosted.org/packages/0b/42/1d16b1f40f0c69d19f4cf0b00014f32f5f5d87ee79f11a0fcd3538520068/session-repository-0.4.5.4.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-01-11 15:15:52",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "Impro02",
"github_project": "session-repository",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"requirements": [],
"lcname": "session-repository"
}
JSON
