| Name | pysql-repo JSON |
| Version |
0.7.4.0
JSON |
| download |
| home_page | https://github.com/Impro02/pysql-repo |
| Summary | A project to have a base repository class to perform select/insert/update/delete with dynamic syntax |
| upload_time | 2024-07-16 09:39:00 |
| maintainer | None |
| docs_url | None |
| author | Maxime MARTIN |
| requires_python | >=3.8 |
| license | MIT |
| keywords |
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
The pysql-repo 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 pysql_repo, 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 pysql-repo
To install pysql-repo, if you already have Python, you can install with:
```
pip install pysql_repo
```
## How to import pysql-repo
To access pysql-repo and its functions import it in your Python code like this:
```
from pysql_repo import Database, Repository, Service, with_session, Operators, LoadingTechnique, RelationshipOption
```
To access pysql-repo and its asyncio functions import it in your Python code like this:
```
from pysql_repo.asyncio import AsyncDatabase, AsyncRepository, AsyncService, with_async_session
```
## Reading the example code
```
# MODULES
from typing import List
# SQLALCHEMY
from sqlalchemy import Boolean, ForeignKey, Integer, String
from sqlalchemy.orm import relationship, Mapped, mapped_column
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class City(Base):
__tablename__ = "CITY"
id: Mapped[int] = mapped_column(
"ID",
Integer,
primary_key=True,
index=True,
)
name: Mapped[str] = mapped_column(
"NAME",
String,
index=True,
)
state: Mapped[str] = mapped_column(
"STATE",
String,
index=True,
)
addresses: Mapped[List["Address"]] = relationship(
"Address",
back_populates="city",
lazy="joined",
cascade="all, delete-orphan",
)
class Address(Base):
__tablename__ = "ADDRESS"
id: Mapped[int] = mapped_column(
"ID",
Integer,
primary_key=True,
index=True,
)
street: Mapped[str] = mapped_column(
"STREET",
String,
index=True,
)
zip_code: Mapped[str] = mapped_column(
"ZIP_CODE",
Integer,
index=True,
)
user_id: Mapped[int] = mapped_column(
"USER_ID",
Integer,
ForeignKey("USER.ID"),
)
city_id: Mapped[int] = mapped_column(
"CITY_ID",
Integer,
ForeignKey("CITY.ID"),
)
user: Mapped["User"] = relationship(
"User",
back_populates="addresses",
lazy="joined",
)
city: Mapped["City"] = relationship(
"City",
back_populates="addresses",
lazy="joined",
)
class User(Base):
__tablename__ = "USER"
id: Mapped[int] = mapped_column(
"ID",
Integer,
primary_key=True,
index=True,
)
email: Mapped[str] = mapped_column(
"EMAIL",
String,
unique=True,
index=True,
)
hashed_password: Mapped[str] = mapped_column(
"HASHED_PASSWORD",
String,
)
full_name: Mapped[str] = mapped_column(
"FULL_NAME",
String,
index=True,
)
is_active: Mapped[bool] = mapped_column(
"IS_ACTIVE",
Boolean,
default=True,
)
addresses: Mapped[List["Address"]] = relationship(
"Address",
back_populates="user",
lazy="joined",
cascade="all, delete-orphan",
)
```
To create an instance of Database and generate an instance of Session from the factory:
```
import logging
logging.basicConfig()
logger_db = logging.get_logger('demo')
database = DataBase(
databases_config={
"connection_string": "foo",
},
base=Base,
logger=logger_db,
)
database.create_database()
with database.session_factory() as session:
session.execute(...)
```
To create a repository, you just have to inherit your class from Repository.
```
# MODULES
from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple
# SQLALCHEMY
from sqlalchemy.orm import Column, Session
# PYSQL_REPO
from pysql_repo import Operators, Repository, LoadingTechnique, RelationshipOption
# CONTEXTLIB
from contextlib import AbstractContextManager
# MODEL
from tests.repositories.user._base import UserRepositoryBase as _UserRepositoryBase
from tests.models.database.database import Address, User
from tests.models.schemas.user import UserCreate
class UserRepositoryBase:
@classmethod
def get_filters(
cls,
ids_in: Optional[List[int]] = None,
ids_not_in: Optional[List[int]] = None,
emails_iin: Optional[List[str]] = None,
emails_in: Optional[List[str]] = None,
emails_not_iin: Optional[List[str]] = None,
emails_not_in: Optional[List[str]] = None,
email_ilike: Optional[List[str]] = None,
email_like: Optional[List[str]] = None,
email_not_ilike: Optional[List[str]] = None,
email_not_like: Optional[List[str]] = None,
email_equal: Optional[str] = None,
email_iequal: Optional[str] = None,
email_different: Optional[str] = None,
email_idifferent: Optional[str] = None,
zip_codes_in: Optional[List[str]] = None,
zip_codes_not_in: Optional[List[str]] = None,
is_active_equal: Optional[bool] = None,
) -> Dict[Column, Any]:
return {
User.id: {
Operators.IN: ids_in,
Operators.NOT_IN: ids_not_in,
},
User.email: {
Operators.IIN: emails_iin,
Operators.IN: emails_in,
Operators.NOT_IIN: emails_not_iin,
Operators.NOT_IN: emails_not_in,
Operators.ILIKE: email_ilike,
Operators.LIKE: email_like,
Operators.NOT_ILIKE: email_not_ilike,
Operators.NOT_LIKE: email_not_like,
Operators.EQUAL: email_equal,
Operators.IEQUAL: email_iequal,
Operators.DIFFERENT: email_different,
Operators.IDIFFERENT: email_idifferent,
},
User.is_active: {
Operators.EQUAL: is_active_equal,
},
User.addresses: {
Operators.ANY: {
Address.zip_code: {
Operators.IN: zip_codes_in,
Operators.NOT_IN: zip_codes_not_in,
},
}
},
}
@classmethod
def get_relationship_options(
cls,
load_addresses: bool = False,
load_city: bool = False,
zip_codes_not_in: Optional[List[int]] = None,
zip_codes_in: Optional[List[int]] = None,
):
extra_join_addresses = []
if zip_codes_not_in:
extra_join_addresses.append(Address.zip_code.not_in(zip_codes_not_in))
if zip_codes_in:
extra_join_addresses.append(Address.zip_code.in_(zip_codes_in))
return {
User.addresses: RelationshipOption(
lazy=LoadingTechnique.JOINED
if load_addresses
else LoadingTechnique.NOLOAD,
added_criteria=extra_join_addresses
if len(extra_join_addresses) > 0
else None,
children={
Address.city: RelationshipOption(
lazy=LoadingTechnique.JOINED
if load_city
else LoadingTechnique.NOLOAD,
)
},
),
}
class UserRepository(Repository, _UserRepositoryBase):
def __init__(
self,
session_factory: Callable[..., AbstractContextManager[Session]],
) -> None:
super().__init__(session_factory)
def get_all(
self,
__session__: Session,
/,
ids_in: Optional[List[int]] = None,
ids_not_in: Optional[List[int]] = None,
emails_iin: Optional[List[str]] = None,
emails_in: Optional[List[str]] = None,
emails_not_iin: Optional[List[str]] = None,
emails_not_in: Optional[List[str]] = None,
email_ilike: Optional[str] = None,
email_like: Optional[str] = None,
email_not_ilike: Optional[str] = None,
email_not_like: Optional[str] = None,
email_equal: Optional[str] = None,
email_iequal: Optional[str] = None,
email_different: Optional[str] = None,
email_idifferent: Optional[str] = None,
zip_codes_in: Optional[List[int]] = None,
zip_codes_not_in: Optional[List[int]] = None,
is_active_equal: Optional[bool] = None,
load_addresses: bool = True,
load_city: bool = True,
order_by: Optional[List[str]] = None,
direction: Optional[List[str]] = None,
) -> Sequence[User]:
users = self._select_all(
__session__,
model=User,
optional_filters=self.get_filters(
ids_in=ids_in,
ids_not_in=ids_not_in,
emails_iin=emails_iin,
emails_in=emails_in,
emails_not_iin=emails_not_iin,
emails_not_in=emails_not_in,
email_ilike=email_ilike,
email_like=email_like,
email_not_ilike=email_not_ilike,
email_not_like=email_not_like,
email_equal=email_equal,
email_iequal=email_iequal,
email_different=email_different,
email_idifferent=email_idifferent,
zip_codes_in=zip_codes_in,
zip_codes_not_in=zip_codes_not_in,
is_active_equal=is_active_equal,
),
relationship_options=self.get_relationship_options(
load_addresses=load_addresses,
load_city=load_city,
zip_codes_in=zip_codes_in,
zip_codes_not_in=zip_codes_not_in,
),
order_by=order_by,
direction=direction,
)
return users
def get_paginate(
self,
__session__: Session,
/,
page: int,
per_page: int,
ids_in: Optional[List[int]] = None,
ids_not_in: Optional[List[int]] = None,
emails_iin: Optional[List[str]] = None,
emails_in: Optional[List[str]] = None,
emails_not_iin: Optional[List[str]] = None,
emails_not_in: Optional[List[str]] = None,
email_ilike: Optional[str] = None,
email_like: Optional[str] = None,
email_not_ilike: Optional[str] = None,
email_not_like: Optional[str] = None,
email_equal: Optional[str] = None,
email_iequal: Optional[str] = None,
email_different: Optional[str] = None,
email_idifferent: Optional[str] = None,
zip_codes_in: Optional[List[int]] = None,
zip_codes_not_in: Optional[List[int]] = None,
is_active_equal: Optional[bool] = None,
load_addresses: bool = True,
load_city: bool = True,
order_by: Optional[List[str]] = None,
direction: Optional[List[str]] = None,
) -> Tuple[Sequence[User], str]:
users, pagination = self._select_paginate(
__session__,
model=User,
optional_filters=self.get_filters(
ids_in=ids_in,
ids_not_in=ids_not_in,
emails_iin=emails_iin,
emails_in=emails_in,
emails_not_iin=emails_not_iin,
emails_not_in=emails_not_in,
email_ilike=email_ilike,
email_like=email_like,
email_not_ilike=email_not_ilike,
email_not_like=email_not_like,
email_equal=email_equal,
email_iequal=email_iequal,
email_different=email_different,
email_idifferent=email_idifferent,
zip_codes_in=zip_codes_in,
zip_codes_not_in=zip_codes_not_in,
is_active_equal=is_active_equal,
),
relationship_options=self.get_relationship_options(
load_addresses=load_addresses,
load_city=load_city,
zip_codes_in=zip_codes_in,
zip_codes_not_in=zip_codes_not_in,
),
order_by=order_by,
direction=direction,
page=page,
per_page=per_page,
)
return users, pagination
def get_by_id(
self,
__session__: Session,
id: int,
) -> Optional[User]:
user = self._select(
__session__,
model=User,
filters={
User.id: {
Operators.EQUAL: id,
},
},
)
return user
def create(
self,
__session__: Session,
/,
data: UserCreate,
flush: bool = False,
commit: bool = True,
) -> User:
user = self._add(
__session__,
model=User,
values={
User.email.key: data.email,
User.hashed_password.key: data.hashed_password,
User.full_name.key: data.full_name,
},
flush=flush,
commit=commit,
)
return user
def create_all(
self,
__session__: Session,
/,
data: List[UserCreate],
flush: bool = False,
commit: bool = True,
) -> Sequence[User]:
users = self._add_all(
__session__,
model=User,
values=[
{
User.email.key: item.email,
User.hashed_password.key: item.hashed_password,
User.full_name.key: item.full_name,
}
for item in data
],
flush=flush,
commit=commit,
)
return users
def patch_email(
self,
__session__: Session,
/,
id: int,
email: str,
flush: bool = False,
commit: bool = True,
) -> Optional[User]:
user = self._update(
__session__,
model=User,
values={
User.email.key: email,
},
filters={
User.id: {
Operators.EQUAL: id,
},
},
flush=flush,
commit=commit,
)
return user
def bulk_patch_email(
self,
__session__: Session,
/,
data: List[Tuple[int, str]],
flush: bool = False,
commit: bool = True,
) -> Sequence[User]:
self._bulk_update(
__session__,
model=User,
values=[
{
User.id.key: id,
User.email.key: email,
}
for id, email in data
],
flush=flush,
commit=commit,
)
return self._select_all(
__session__,
model=User,
filters={
User.id: {
Operators.IN: [id for id, _ in data],
},
},
)
def patch_disable(
self,
__session__: Session,
/,
ids: List[int],
flush: bool = False,
commit: bool = True,
) -> Sequence[User]:
users = self._update_all(
__session__,
model=User,
values={
User.is_active.key: False,
},
filters={
User.id: {
Operators.IN: ids,
},
},
flush=flush,
commit=commit,
)
return users
def delete(
self,
__session__: Session,
/,
id: int,
flush: bool = False,
commit: bool = True,
) -> bool:
is_deleted = self._delete(
__session__,
model=User,
filters={
User.id: {
Operators.EQUAL: id,
},
},
flush=flush,
commit=commit,
)
return is_deleted
def delete_all(
self,
__session__: Session,
/,
ids: List[int],
flush: bool = False,
commit: bool = True,
) -> bool:
is_deleted = self._delete_all(
__session__,
model=User,
filters={
User.id: {
Operators.IN: ids,
},
},
flush=flush,
commit=commit,
)
return is_deleted
```
To create a service, you just have to inherit your class from Service.
```
# MODULES
from logging import Logger
from typing import List, Optional, Tuple
# SQLALCHEMY
from sqlalchemy.orm import Session
# PYSQL_REPO
from pysql_repo import Service, with_session
# REPOSITORIES
from tests.repositories.user.user_repository import UserRepository
# MODELS
from tests.models.schemas.user import UserCreate, UserRead
class UserService(Service[UserRepository]):
def __init__(
self,
user_repository: UserRepository,
logger: Logger,
) -> None:
super().__init__(
repository=user_repository,
)
self._logger = logger
def get_users(
self,
__session__: Session,
/,
ids_in: Optional[List[int]] = None,
ids_not_in: Optional[List[int]] = None,
emails_iin: Optional[List[str]] = None,
emails_in: Optional[List[str]] = None,
emails_not_iin: Optional[List[str]] = None,
emails_not_in: Optional[List[str]] = None,
email_ilike: Optional[str] = None,
email_like: Optional[str] = None,
email_not_ilike: Optional[str] = None,
email_not_like: Optional[str] = None,
email_equal: Optional[str] = None,
email_iequal: Optional[str] = None,
email_different: Optional[str] = None,
email_idifferent: Optional[str] = None,
zip_codes_in: Optional[List[int]] = None,
zip_codes_not_in: Optional[List[int]] = None,
is_active_equal: Optional[bool] = None,
load_addresses: bool = True,
load_city: bool = True,
order_by: Optional[List[str]] = None,
direction: Optional[List[str]] = None,
) -> List[UserRead]:
users = self._repository.get_all(
__session__,
ids_in=ids_in,
ids_not_in=ids_not_in,
emails_iin=emails_iin,
emails_in=emails_in,
emails_not_iin=emails_not_iin,
emails_not_in=emails_not_in,
email_ilike=email_ilike,
email_like=email_like,
email_not_ilike=email_not_ilike,
email_not_like=email_not_like,
email_equal=email_equal,
email_iequal=email_iequal,
email_different=email_different,
email_idifferent=email_idifferent,
zip_codes_in=zip_codes_in,
zip_codes_not_in=zip_codes_not_in,
is_active_equal=is_active_equal,
load_addresses=load_addresses,
load_city=load_city,
order_by=order_by,
direction=direction,
)
return [UserRead.model_validate(item) for item in users]
def get_users_paginate(
self,
__session__: Session,
/,
page: int,
per_page: int,
ids_in: Optional[List[int]] = None,
ids_not_in: Optional[List[int]] = None,
emails_iin: Optional[List[str]] = None,
emails_in: Optional[List[str]] = None,
emails_not_iin: Optional[List[str]] = None,
emails_not_in: Optional[List[str]] = None,
email_ilike: Optional[str] = None,
email_like: Optional[str] = None,
email_not_ilike: Optional[str] = None,
email_not_like: Optional[str] = None,
email_equal: Optional[str] = None,
email_iequal: Optional[str] = None,
email_different: Optional[str] = None,
email_idifferent: Optional[str] = None,
zip_codes_in: Optional[List[int]] = None,
zip_codes_not_in: Optional[List[int]] = None,
is_active_equal: Optional[bool] = None,
load_addresses: bool = True,
load_city: bool = True,
order_by: Optional[List[str]] = None,
direction: Optional[List[str]] = None,
) -> Tuple[List[UserRead], str]:
users, pagination = self._repository.get_paginate(
__session__,
page=page,
per_page=per_page,
ids_in=ids_in,
ids_not_in=ids_not_in,
emails_iin=emails_iin,
emails_in=emails_in,
emails_not_iin=emails_not_iin,
emails_not_in=emails_not_in,
email_ilike=email_ilike,
email_like=email_like,
email_not_ilike=email_not_ilike,
email_not_like=email_not_like,
email_equal=email_equal,
email_iequal=email_iequal,
email_different=email_different,
email_idifferent=email_idifferent,
zip_codes_in=zip_codes_in,
zip_codes_not_in=zip_codes_not_in,
is_active_equal=is_active_equal,
load_addresses=load_addresses,
load_city=load_city,
order_by=order_by,
direction=direction,
)
return [UserRead.model_validate(item) for item in users], pagination
def get_user_by_id(
self,
__session__: Session,
/,
id: int,
) -> Optional[UserRead]:
user = self._repository.get_by_id(
__session__,
id=id,
)
if user is None:
return None
return UserRead.model_validate(user)
def create_user(
self,
__session__: Session,
/,
data: UserCreate,
) -> UserRead:
user = self._repository.create(
__session__,
data=data,
flush=True,
)
return UserRead.model_validate(user)
def create_users(
self,
__session__: Session,
/,
data: List[UserCreate],
) -> List[UserRead]:
users = self._repository.create_all(
__session__,
data=data,
flush=True,
)
return [UserRead.model_validate(user) for user in users]
def patch_email(
self,
__session__: Session,
/,
id: int,
email: str,
) -> UserRead:
user = self._repository.patch_email(
__session__,
id=id,
email=email,
flush=True,
)
return UserRead.model_validate(user)
def bulk_patch_email(
self,
__session__: Session,
/,
data: List[Tuple[int, str]],
) -> List[UserRead]:
users = self._repository.bulk_patch_email(
__session__,
data=data,
flush=True,
)
return [UserRead.model_validate(user) for user in users]
def patch_disable(
self,
__session__: Session,
/,
ids: List[int],
) -> List[UserRead]:
users = self._repository.patch_disable(
__session__,
ids=ids,
flush=True,
)
return [UserRead.model_validate(user) for user in users]
def delete_by_id(
self,
__session__: Session,
/,
id: int,
) -> bool:
return self._repository.delete(
__session__,
id=id,
flush=True,
)
def delete_by_ids(
self,
__session__: Session,
/,
ids: List[int],
) -> bool:
return self._repository.delete_all(
__session__,
ids=ids,
flush=True,
)
```
To create an instance of AsyncDatabase and generate an instance of AsyncSession from the factory:
```
import logging
logging.basicConfig()
logger_db = logging.get_logger('demo')
database = AsyncDataBase(
databases_config={
"connection_string": "foo",
},
base=Base,
logger=logger_db,
)
await database.create_database()
async with database.session_factory() as session:
await session.execute(...)
```
To create an async repository, you just have to inherit your class from AsyncRepository.
```
# MODULES
from typing import Callable, List, Optional, Tuple
# SQLALCHEMY
from sqlalchemy import Sequence
from sqlalchemy.ext.asyncio import AsyncSession
# PYSQL_REPO
from pysql_repo import Operators
from pysql_repo.asyncio import AsyncRepository
# CONTEXTLIB
from contextlib import AbstractAsyncContextManager
# MODEL
from tests.repositories.user._base import UserRepositoryBase as _UserRepositoryBase
from tests.models.database.database import User
from tests.models.schemas.user import UserCreate
class AsyncUserRepository(AsyncRepository, _UserRepositoryBase):
def __init__(
self,
session_factory: Callable[..., AbstractAsyncContextManager[AsyncSession]],
) -> None:
super().__init__(session_factory)
async def get_all(
self,
__session__: AsyncSession,
/,
ids_in: Optional[List[int]] = None,
ids_not_in: Optional[List[int]] = None,
emails_iin: Optional[List[str]] = None,
emails_in: Optional[List[str]] = None,
emails_not_iin: Optional[List[str]] = None,
emails_not_in: Optional[List[str]] = None,
email_ilike: Optional[str] = None,
email_like: Optional[str] = None,
email_not_ilike: Optional[str] = None,
email_not_like: Optional[str] = None,
email_equal: Optional[str] = None,
email_iequal: Optional[str] = None,
email_different: Optional[str] = None,
email_idifferent: Optional[str] = None,
zip_codes_in: Optional[List[int]] = None,
zip_codes_not_in: Optional[List[int]] = None,
is_active_equal: Optional[bool] = None,
load_addresses: bool = True,
load_city: bool = True,
order_by: Optional[List[str]] = None,
direction: Optional[List[str]] = None,
) -> Sequence[User]:
users = await self._select_all(
__session__,
model=User,
optional_filters=self.get_filters(
ids_in=ids_in,
ids_not_in=ids_not_in,
emails_iin=emails_iin,
emails_in=emails_in,
emails_not_iin=emails_not_iin,
emails_not_in=emails_not_in,
email_ilike=email_ilike,
email_like=email_like,
email_not_ilike=email_not_ilike,
email_not_like=email_not_like,
email_equal=email_equal,
email_iequal=email_iequal,
email_different=email_different,
email_idifferent=email_idifferent,
zip_codes_in=zip_codes_in,
zip_codes_not_in=zip_codes_not_in,
is_active_equal=is_active_equal,
),
relationship_options=self.get_relationship_options(
load_addresses=load_addresses,
load_city=load_city,
zip_codes_in=zip_codes_in,
zip_codes_not_in=zip_codes_not_in,
),
order_by=order_by,
direction=direction,
)
return users
async def get_paginate(
self,
__session__: AsyncSession,
/,
page: int,
per_page: int,
ids_in: Optional[List[int]] = None,
ids_not_in: Optional[List[int]] = None,
emails_iin: Optional[List[str]] = None,
emails_in: Optional[List[str]] = None,
emails_not_iin: Optional[List[str]] = None,
emails_not_in: Optional[List[str]] = None,
email_ilike: Optional[str] = None,
email_like: Optional[str] = None,
email_not_ilike: Optional[str] = None,
email_not_like: Optional[str] = None,
email_equal: Optional[str] = None,
email_iequal: Optional[str] = None,
email_different: Optional[str] = None,
email_idifferent: Optional[str] = None,
zip_codes_in: Optional[List[int]] = None,
zip_codes_not_in: Optional[List[int]] = None,
is_active_equal: Optional[bool] = None,
load_addresses: bool = True,
load_city: bool = True,
order_by: Optional[List[str]] = None,
direction: Optional[List[str]] = None,
) -> Tuple[Sequence[User], str]:
users, pagination = await self._select_paginate(
__session__,
model=User,
optional_filters=self.get_filters(
ids_in=ids_in,
ids_not_in=ids_not_in,
emails_iin=emails_iin,
emails_in=emails_in,
emails_not_iin=emails_not_iin,
emails_not_in=emails_not_in,
email_ilike=email_ilike,
email_like=email_like,
email_not_ilike=email_not_ilike,
email_not_like=email_not_like,
email_equal=email_equal,
email_iequal=email_iequal,
email_different=email_different,
email_idifferent=email_idifferent,
zip_codes_in=zip_codes_in,
zip_codes_not_in=zip_codes_not_in,
is_active_equal=is_active_equal,
),
relationship_options=self.get_relationship_options(
load_addresses=load_addresses,
load_city=load_city,
zip_codes_in=zip_codes_in,
zip_codes_not_in=zip_codes_not_in,
),
order_by=order_by,
direction=direction,
page=page,
per_page=per_page,
)
return users, pagination
async def get_by_id(
self,
__session__: AsyncSession,
/,
id: int,
) -> Optional[User]:
user = await self._select(
__session__,
model=User,
filters={
User.id: {
Operators.EQUAL: id,
},
},
)
return user
async def create(
self,
__session__: AsyncSession,
/,
data: UserCreate,
flush: bool = False,
commit: bool = True,
) -> User:
user = await self._add(
__session__,
model=User,
values={
User.email.key: data.email,
User.hashed_password.key: data.hashed_password,
User.full_name.key: data.full_name,
},
flush=flush,
commit=commit,
)
return user
async def create_all(
self,
__session__: AsyncSession,
/,
data: List[UserCreate],
flush: bool = False,
commit: bool = True,
) -> Sequence[User]:
users = await self._add_all(
__session__,
model=User,
values=[
{
User.email.key: item.email,
User.hashed_password.key: item.hashed_password,
User.full_name.key: item.full_name,
}
for item in data
],
flush=flush,
commit=commit,
)
return users
async def patch_email(
self,
__session__: AsyncSession,
/,
id: int,
email: str,
flush: bool = False,
commit: bool = True,
) -> Optional[User]:
user = await self._update(
__session__,
model=User,
values={
User.email.key: email,
},
filters={
User.id: {
Operators.EQUAL: id,
},
},
flush=flush,
commit=commit,
)
return user
async def bulk_patch_email(
self,
__session__: AsyncSession,
/,
data: List[Tuple[int, str]],
flush: bool = False,
commit: bool = True,
) -> Sequence[User]:
await self._bulk_update(
__session__,
model=User,
values=[
{
User.id.key: id,
User.email.key: email,
}
for id, email in data
],
flush=flush,
commit=commit,
)
return await self._select_all(
__session__,
model=User,
filters={
User.id: {
Operators.IN: [id for id, _ in data],
},
},
)
async def patch_disable(
self,
__session__: AsyncSession,
/,
ids: List[int],
flush: bool = False,
commit: bool = True,
) -> Sequence[User]:
users = await self._update_all(
__session__,
model=User,
values={
User.is_active.key: False,
},
filters={
User.id: {
Operators.IN: ids,
},
},
flush=flush,
commit=commit,
)
return users
async def delete(
self,
__session__: AsyncSession,
/,
id: int,
flush: bool = False,
commit: bool = True,
) -> bool:
is_deleted = await self._delete(
__session__,
model=User,
filters={
User.id: {
Operators.EQUAL: id,
},
},
flush=flush,
commit=commit,
)
return is_deleted
async def delete_all(
self,
__session__: AsyncSession,
/,
ids: List[int],
flush: bool = False,
commit: bool = True,
) -> bool:
is_deleted = await self._delete_all(
__session__,
model=User,
filters={
User.id: {
Operators.IN: ids,
},
},
flush=flush,
commit=commit,
)
return is_deleted
```
To create an async service, you just have to inherit your class from AsyncService.
```
# MODULES
from logging import Logger
from typing import List, Optional, Tuple
# SQLALCHEMY
from sqlalchemy.ext.asyncio import AsyncSession
# PYSQL_REPO
from pysql_repo.asyncio import AsyncService, with_async_session
# REPOSITORIES
from tests.repositories.user.async_user_repository import AsyncUserRepository
# MODELS
from tests.models.schemas.user import UserCreate, UserRead
class AsyncUserService(AsyncService[AsyncUserRepository]):
def __init__(
self,
user_repository: AsyncUserRepository,
logger: Logger,
) -> None:
super().__init__(
repository=user_repository,
)
self._logger = logger
async def get_users(
self,
__session__: AsyncSession,
/,
ids_in: Optional[List[int]] = None,
ids_not_in: Optional[List[int]] = None,
emails_iin: Optional[List[str]] = None,
emails_in: Optional[List[str]] = None,
emails_not_iin: Optional[List[str]] = None,
emails_not_in: Optional[List[str]] = None,
email_ilike: Optional[str] = None,
email_like: Optional[str] = None,
email_not_ilike: Optional[str] = None,
email_not_like: Optional[str] = None,
email_equal: Optional[str] = None,
email_iequal: Optional[str] = None,
email_different: Optional[str] = None,
email_idifferent: Optional[str] = None,
zip_codes_in: Optional[List[int]] = None,
zip_codes_not_in: Optional[List[int]] = None,
is_active_equal: Optional[bool] = None,
load_addresses: bool = True,
load_city: bool = True,
order_by: Optional[List[str]] = None,
direction: Optional[List[str]] = None,
) -> List[UserRead]:
users = await self._repository.get_all(
__session__,
ids_in=ids_in,
ids_not_in=ids_not_in,
emails_iin=emails_iin,
emails_in=emails_in,
emails_not_iin=emails_not_iin,
emails_not_in=emails_not_in,
email_ilike=email_ilike,
email_like=email_like,
email_not_ilike=email_not_ilike,
email_not_like=email_not_like,
email_equal=email_equal,
email_iequal=email_iequal,
email_different=email_different,
email_idifferent=email_idifferent,
zip_codes_in=zip_codes_in,
zip_codes_not_in=zip_codes_not_in,
is_active_equal=is_active_equal,
load_addresses=load_addresses,
load_city=load_city,
order_by=order_by,
direction=direction,
)
return [UserRead.model_validate(item) for item in users]
async def get_users_paginate(
self,
__session__: AsyncSession,
/,
page: int,
per_page: int,
ids_in: Optional[List[int]] = None,
ids_not_in: Optional[List[int]] = None,
emails_iin: Optional[List[str]] = None,
emails_in: Optional[List[str]] = None,
emails_not_iin: Optional[List[str]] = None,
emails_not_in: Optional[List[str]] = None,
email_ilike: Optional[str] = None,
email_like: Optional[str] = None,
email_not_ilike: Optional[str] = None,
email_not_like: Optional[str] = None,
email_equal: Optional[str] = None,
email_iequal: Optional[str] = None,
email_different: Optional[str] = None,
email_idifferent: Optional[str] = None,
zip_codes_in: Optional[List[int]] = None,
zip_codes_not_in: Optional[List[int]] = None,
is_active_equal: Optional[bool] = None,
load_addresses: bool = True,
load_city: bool = True,
order_by: Optional[List[str]] = None,
direction: Optional[List[str]] = None,
) -> Tuple[List[UserRead], str]:
users, pagination = await self._repository.get_paginate(
__session__,
page=page,
per_page=per_page,
ids_in=ids_in,
ids_not_in=ids_not_in,
emails_iin=emails_iin,
emails_in=emails_in,
emails_not_iin=emails_not_iin,
emails_not_in=emails_not_in,
email_ilike=email_ilike,
email_like=email_like,
email_not_ilike=email_not_ilike,
email_not_like=email_not_like,
email_equal=email_equal,
email_iequal=email_iequal,
email_different=email_different,
email_idifferent=email_idifferent,
zip_codes_in=zip_codes_in,
zip_codes_not_in=zip_codes_not_in,
is_active_equal=is_active_equal,
load_addresses=load_addresses,
load_city=load_city,
order_by=order_by,
direction=direction,
)
return [UserRead.model_validate(item) for item in users], pagination
async def get_user_by_id(
self,
__session__: AsyncSession,
/,
id: int,
) -> Optional[UserRead]:
user = await self._repository.get_by_id(
__session__,
id=id,
)
if user is None:
return None
return UserRead.model_validate(user)
async def create_user(
self,
__session__: AsyncSession,
/,
data: UserCreate,
) -> UserRead:
user = await self._repository.create(
__session__,
data=data,
flush=True,
)
return UserRead.model_validate(user)
async def create_users(
self,
__session__: AsyncSession,
/,
data: List[UserCreate],
) -> List[UserRead]:
users = await self._repository.create_all(
__session__,
data=data,
flush=True,
)
return [UserRead.model_validate(user) for user in users]
async def patch_email(
self,
__session__: AsyncSession,
/,
id: int,
email: str,
) -> UserRead:
user = await self._repository.patch_email(
__session__,
id=id,
email=email,
flush=True,
)
return UserRead.model_validate(user)
async def bulk_patch_email(
self,
__session__: AsyncSession,
/,
data: List[Tuple[int, str]],
) -> List[UserRead]:
users = await self._repository.bulk_patch_email(
__session__,
data=data,
flush=True,
)
return [UserRead.model_validate(user) for user in users]
async def patch_disable(
self,
__session__: AsyncSession,
/,
ids: List[int],
) -> List[UserRead]:
users = await self._repository.patch_disable(
__session__,
ids=ids,
flush=True,
)
return [UserRead.model_validate(user) for user in users]
async def delete_by_id(
self,
__session__: AsyncSession,
/,
id: int,
) -> bool:
return await self._repository.delete(
__session__,
id=id,
flush=True,
)
async def delete_by_ids(
self,
__session__: AsyncSession,
/,
ids: List[int],
) -> bool:
return await self._repository.delete_all(
__session__,
ids=ids,
flush=True,
)
```
Raw data
{
"_id": null,
"home_page": "https://github.com/Impro02/pysql-repo",
"name": "pysql-repo",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": null,
"keywords": null,
"author": "Maxime MARTIN",
"author_email": "Maxime MARTIN <maxime.martin02@hotmail.fr>",
"download_url": "https://files.pythonhosted.org/packages/48/19/bacf94b17a2352cf8dd736da1b6e6ff1170aa584787031380bf816a8ca28/pysql-repo-0.7.4.0.tar.gz",
"platform": null,
"description": "The pysql-repo 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 pysql_repo, 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 pysql-repo\n\nTo install pysql-repo, if you already have Python, you can install with:\n\n```\npip install pysql_repo\n```\n\n## How to import pysql-repo\n\nTo access pysql-repo and its functions import it in your Python code like this:\n\n```\nfrom pysql_repo import Database, Repository, Service, with_session, Operators, LoadingTechnique, RelationshipOption\n```\n\nTo access pysql-repo and its asyncio functions import it in your Python code like this:\n\n```\nfrom pysql_repo.asyncio import AsyncDatabase, AsyncRepository, AsyncService, with_async_session\n```\n\n## Reading the example code\n\n```\n# MODULES\nfrom typing import List\n\n# SQLALCHEMY\nfrom sqlalchemy import Boolean, ForeignKey, Integer, String\nfrom sqlalchemy.orm import relationship, Mapped, mapped_column\nfrom sqlalchemy.ext.declarative import declarative_base\n\nBase = declarative_base()\n\n\nclass City(Base):\n __tablename__ = \"CITY\"\n\n id: Mapped[int] = mapped_column(\n \"ID\",\n Integer,\n primary_key=True,\n index=True,\n )\n name: Mapped[str] = mapped_column(\n \"NAME\",\n String,\n index=True,\n )\n state: Mapped[str] = mapped_column(\n \"STATE\",\n String,\n index=True,\n )\n\n addresses: Mapped[List[\"Address\"]] = relationship(\n \"Address\",\n back_populates=\"city\",\n lazy=\"joined\",\n cascade=\"all, delete-orphan\",\n )\n\n\nclass Address(Base):\n __tablename__ = \"ADDRESS\"\n\n id: Mapped[int] = mapped_column(\n \"ID\",\n Integer,\n primary_key=True,\n index=True,\n )\n street: Mapped[str] = mapped_column(\n \"STREET\",\n String,\n index=True,\n )\n zip_code: Mapped[str] = mapped_column(\n \"ZIP_CODE\",\n Integer,\n index=True,\n )\n user_id: Mapped[int] = mapped_column(\n \"USER_ID\",\n Integer,\n ForeignKey(\"USER.ID\"),\n )\n city_id: Mapped[int] = mapped_column(\n \"CITY_ID\",\n Integer,\n ForeignKey(\"CITY.ID\"),\n )\n\n user: Mapped[\"User\"] = relationship(\n \"User\",\n back_populates=\"addresses\",\n lazy=\"joined\",\n )\n city: Mapped[\"City\"] = relationship(\n \"City\",\n back_populates=\"addresses\",\n lazy=\"joined\",\n )\n\n\nclass User(Base):\n __tablename__ = \"USER\"\n\n id: Mapped[int] = mapped_column(\n \"ID\",\n Integer,\n primary_key=True,\n index=True,\n )\n email: Mapped[str] = mapped_column(\n \"EMAIL\",\n String,\n unique=True,\n index=True,\n )\n hashed_password: Mapped[str] = mapped_column(\n \"HASHED_PASSWORD\",\n String,\n )\n full_name: Mapped[str] = mapped_column(\n \"FULL_NAME\",\n String,\n index=True,\n )\n is_active: Mapped[bool] = mapped_column(\n \"IS_ACTIVE\",\n Boolean,\n default=True,\n )\n\n addresses: Mapped[List[\"Address\"]] = relationship(\n \"Address\",\n back_populates=\"user\",\n lazy=\"joined\",\n cascade=\"all, delete-orphan\",\n )\n```\n\nTo create an instance of Database and generate an instance of Session from the factory:\n\n```\nimport logging\n\nlogging.basicConfig()\nlogger_db = logging.get_logger('demo')\n\ndatabase = DataBase(\n databases_config={\n \"connection_string\": \"foo\",\n },\n base=Base,\n logger=logger_db,\n)\n\ndatabase.create_database()\n\nwith database.session_factory() as session:\n session.execute(...)\n```\n\nTo create a repository, you just have to inherit your class from Repository.\n\n```\n# MODULES\nfrom typing import Any, Callable, Dict, List, Optional, Sequence, Tuple\n\n# SQLALCHEMY\nfrom sqlalchemy.orm import Column, Session\n\n# PYSQL_REPO\nfrom pysql_repo import Operators, Repository, LoadingTechnique, RelationshipOption\n\n# CONTEXTLIB\nfrom contextlib import AbstractContextManager\n\n# MODEL\nfrom tests.repositories.user._base import UserRepositoryBase as _UserRepositoryBase\nfrom tests.models.database.database import Address, User\nfrom tests.models.schemas.user import UserCreate\n\n\nclass UserRepositoryBase:\n @classmethod\n def get_filters(\n cls,\n ids_in: Optional[List[int]] = None,\n ids_not_in: Optional[List[int]] = None,\n emails_iin: Optional[List[str]] = None,\n emails_in: Optional[List[str]] = None,\n emails_not_iin: Optional[List[str]] = None,\n emails_not_in: Optional[List[str]] = None,\n email_ilike: Optional[List[str]] = None,\n email_like: Optional[List[str]] = None,\n email_not_ilike: Optional[List[str]] = None,\n email_not_like: Optional[List[str]] = None,\n email_equal: Optional[str] = None,\n email_iequal: Optional[str] = None,\n email_different: Optional[str] = None,\n email_idifferent: Optional[str] = None,\n zip_codes_in: Optional[List[str]] = None,\n zip_codes_not_in: Optional[List[str]] = None,\n is_active_equal: Optional[bool] = None,\n ) -> Dict[Column, Any]:\n return {\n User.id: {\n Operators.IN: ids_in,\n Operators.NOT_IN: ids_not_in,\n },\n User.email: {\n Operators.IIN: emails_iin,\n Operators.IN: emails_in,\n Operators.NOT_IIN: emails_not_iin,\n Operators.NOT_IN: emails_not_in,\n Operators.ILIKE: email_ilike,\n Operators.LIKE: email_like,\n Operators.NOT_ILIKE: email_not_ilike,\n Operators.NOT_LIKE: email_not_like,\n Operators.EQUAL: email_equal,\n Operators.IEQUAL: email_iequal,\n Operators.DIFFERENT: email_different,\n Operators.IDIFFERENT: email_idifferent,\n },\n User.is_active: {\n Operators.EQUAL: is_active_equal,\n },\n User.addresses: {\n Operators.ANY: {\n Address.zip_code: {\n Operators.IN: zip_codes_in,\n Operators.NOT_IN: zip_codes_not_in,\n },\n }\n },\n }\n\n @classmethod\n def get_relationship_options(\n cls,\n load_addresses: bool = False,\n load_city: bool = False,\n zip_codes_not_in: Optional[List[int]] = None,\n zip_codes_in: Optional[List[int]] = None,\n ):\n extra_join_addresses = []\n if zip_codes_not_in:\n extra_join_addresses.append(Address.zip_code.not_in(zip_codes_not_in))\n if zip_codes_in:\n extra_join_addresses.append(Address.zip_code.in_(zip_codes_in))\n\n return {\n User.addresses: RelationshipOption(\n lazy=LoadingTechnique.JOINED\n if load_addresses\n else LoadingTechnique.NOLOAD,\n added_criteria=extra_join_addresses\n if len(extra_join_addresses) > 0\n else None,\n children={\n Address.city: RelationshipOption(\n lazy=LoadingTechnique.JOINED\n if load_city\n else LoadingTechnique.NOLOAD,\n )\n },\n ),\n }\n\n\nclass UserRepository(Repository, _UserRepositoryBase):\n def __init__(\n self,\n session_factory: Callable[..., AbstractContextManager[Session]],\n ) -> None:\n super().__init__(session_factory)\n\n def get_all(\n self,\n __session__: Session,\n /,\n ids_in: Optional[List[int]] = None,\n ids_not_in: Optional[List[int]] = None,\n emails_iin: Optional[List[str]] = None,\n emails_in: Optional[List[str]] = None,\n emails_not_iin: Optional[List[str]] = None,\n emails_not_in: Optional[List[str]] = None,\n email_ilike: Optional[str] = None,\n email_like: Optional[str] = None,\n email_not_ilike: Optional[str] = None,\n email_not_like: Optional[str] = None,\n email_equal: Optional[str] = None,\n email_iequal: Optional[str] = None,\n email_different: Optional[str] = None,\n email_idifferent: Optional[str] = None,\n zip_codes_in: Optional[List[int]] = None,\n zip_codes_not_in: Optional[List[int]] = None,\n is_active_equal: Optional[bool] = None,\n load_addresses: bool = True,\n load_city: bool = True,\n order_by: Optional[List[str]] = None,\n direction: Optional[List[str]] = None,\n ) -> Sequence[User]:\n users = self._select_all(\n __session__,\n model=User,\n optional_filters=self.get_filters(\n ids_in=ids_in,\n ids_not_in=ids_not_in,\n emails_iin=emails_iin,\n emails_in=emails_in,\n emails_not_iin=emails_not_iin,\n emails_not_in=emails_not_in,\n email_ilike=email_ilike,\n email_like=email_like,\n email_not_ilike=email_not_ilike,\n email_not_like=email_not_like,\n email_equal=email_equal,\n email_iequal=email_iequal,\n email_different=email_different,\n email_idifferent=email_idifferent,\n zip_codes_in=zip_codes_in,\n zip_codes_not_in=zip_codes_not_in,\n is_active_equal=is_active_equal,\n ),\n relationship_options=self.get_relationship_options(\n load_addresses=load_addresses,\n load_city=load_city,\n zip_codes_in=zip_codes_in,\n zip_codes_not_in=zip_codes_not_in,\n ),\n order_by=order_by,\n direction=direction,\n )\n\n return users\n\n def get_paginate(\n self,\n __session__: Session,\n /,\n page: int,\n per_page: int,\n ids_in: Optional[List[int]] = None,\n ids_not_in: Optional[List[int]] = None,\n emails_iin: Optional[List[str]] = None,\n emails_in: Optional[List[str]] = None,\n emails_not_iin: Optional[List[str]] = None,\n emails_not_in: Optional[List[str]] = None,\n email_ilike: Optional[str] = None,\n email_like: Optional[str] = None,\n email_not_ilike: Optional[str] = None,\n email_not_like: Optional[str] = None,\n email_equal: Optional[str] = None,\n email_iequal: Optional[str] = None,\n email_different: Optional[str] = None,\n email_idifferent: Optional[str] = None,\n zip_codes_in: Optional[List[int]] = None,\n zip_codes_not_in: Optional[List[int]] = None,\n is_active_equal: Optional[bool] = None,\n load_addresses: bool = True,\n load_city: bool = True,\n order_by: Optional[List[str]] = None,\n direction: Optional[List[str]] = None,\n ) -> Tuple[Sequence[User], str]:\n users, pagination = self._select_paginate(\n __session__,\n model=User,\n optional_filters=self.get_filters(\n ids_in=ids_in,\n ids_not_in=ids_not_in,\n emails_iin=emails_iin,\n emails_in=emails_in,\n emails_not_iin=emails_not_iin,\n emails_not_in=emails_not_in,\n email_ilike=email_ilike,\n email_like=email_like,\n email_not_ilike=email_not_ilike,\n email_not_like=email_not_like,\n email_equal=email_equal,\n email_iequal=email_iequal,\n email_different=email_different,\n email_idifferent=email_idifferent,\n zip_codes_in=zip_codes_in,\n zip_codes_not_in=zip_codes_not_in,\n is_active_equal=is_active_equal,\n ),\n relationship_options=self.get_relationship_options(\n load_addresses=load_addresses,\n load_city=load_city,\n zip_codes_in=zip_codes_in,\n zip_codes_not_in=zip_codes_not_in,\n ),\n order_by=order_by,\n direction=direction,\n page=page,\n per_page=per_page,\n )\n\n return users, pagination\n\n def get_by_id(\n self,\n __session__: Session,\n id: int,\n ) -> Optional[User]:\n user = self._select(\n __session__,\n model=User,\n filters={\n User.id: {\n Operators.EQUAL: id,\n },\n },\n )\n\n return user\n\n def create(\n self,\n __session__: Session,\n /,\n data: UserCreate,\n flush: bool = False,\n commit: bool = True,\n ) -> User:\n user = self._add(\n __session__,\n model=User,\n values={\n User.email.key: data.email,\n User.hashed_password.key: data.hashed_password,\n User.full_name.key: data.full_name,\n },\n flush=flush,\n commit=commit,\n )\n\n return user\n\n def create_all(\n self,\n __session__: Session,\n /,\n data: List[UserCreate],\n flush: bool = False,\n commit: bool = True,\n ) -> Sequence[User]:\n users = self._add_all(\n __session__,\n model=User,\n values=[\n {\n User.email.key: item.email,\n User.hashed_password.key: item.hashed_password,\n User.full_name.key: item.full_name,\n }\n for item in data\n ],\n flush=flush,\n commit=commit,\n )\n\n return users\n\n def patch_email(\n self,\n __session__: Session,\n /,\n id: int,\n email: str,\n flush: bool = False,\n commit: bool = True,\n ) -> Optional[User]:\n user = self._update(\n __session__,\n model=User,\n values={\n User.email.key: email,\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 bulk_patch_email(\n self,\n __session__: Session,\n /,\n data: List[Tuple[int, str]],\n flush: bool = False,\n commit: bool = True,\n ) -> Sequence[User]:\n self._bulk_update(\n __session__,\n model=User,\n values=[\n {\n User.id.key: id,\n User.email.key: email,\n }\n for id, email in data\n ],\n flush=flush,\n commit=commit,\n )\n\n return self._select_all(\n __session__,\n model=User,\n filters={\n User.id: {\n Operators.IN: [id for id, _ in data],\n },\n },\n )\n\n def patch_disable(\n self,\n __session__: Session,\n /,\n ids: List[int],\n flush: bool = False,\n commit: bool = True,\n ) -> Sequence[User]:\n users = self._update_all(\n __session__,\n model=User,\n values={\n User.is_active.key: False,\n },\n filters={\n User.id: {\n Operators.IN: ids,\n },\n },\n flush=flush,\n commit=commit,\n )\n\n return users\n\n def delete(\n self,\n __session__: Session,\n /,\n id: int,\n flush: bool = False,\n commit: bool = True,\n ) -> bool:\n is_deleted = self._delete(\n __session__,\n model=User,\n filters={\n User.id: {\n Operators.EQUAL: id,\n },\n },\n flush=flush,\n commit=commit,\n )\n\n return is_deleted\n\n def delete_all(\n self,\n __session__: Session,\n /,\n ids: List[int],\n flush: bool = False,\n commit: bool = True,\n ) -> bool:\n is_deleted = self._delete_all(\n __session__,\n model=User,\n filters={\n User.id: {\n Operators.IN: ids,\n },\n },\n flush=flush,\n commit=commit,\n )\n\n return is_deleted\n```\n\nTo create a service, you just have to inherit your class from Service.\n\n```\n# MODULES\nfrom logging import Logger\nfrom typing import List, Optional, Tuple\n\n# SQLALCHEMY\nfrom sqlalchemy.orm import Session\n\n# PYSQL_REPO\nfrom pysql_repo import Service, with_session\n\n# REPOSITORIES\nfrom tests.repositories.user.user_repository import UserRepository\n\n# MODELS\nfrom tests.models.schemas.user import UserCreate, UserRead\n\n\nclass UserService(Service[UserRepository]):\n def __init__(\n self,\n user_repository: UserRepository,\n logger: Logger,\n ) -> None:\n super().__init__(\n repository=user_repository,\n )\n self._logger = logger\n\n def get_users(\n self,\n __session__: Session,\n /,\n ids_in: Optional[List[int]] = None,\n ids_not_in: Optional[List[int]] = None,\n emails_iin: Optional[List[str]] = None,\n emails_in: Optional[List[str]] = None,\n emails_not_iin: Optional[List[str]] = None,\n emails_not_in: Optional[List[str]] = None,\n email_ilike: Optional[str] = None,\n email_like: Optional[str] = None,\n email_not_ilike: Optional[str] = None,\n email_not_like: Optional[str] = None,\n email_equal: Optional[str] = None,\n email_iequal: Optional[str] = None,\n email_different: Optional[str] = None,\n email_idifferent: Optional[str] = None,\n zip_codes_in: Optional[List[int]] = None,\n zip_codes_not_in: Optional[List[int]] = None,\n is_active_equal: Optional[bool] = None,\n load_addresses: bool = True,\n load_city: bool = True,\n order_by: Optional[List[str]] = None,\n direction: Optional[List[str]] = None,\n ) -> List[UserRead]:\n users = self._repository.get_all(\n __session__,\n ids_in=ids_in,\n ids_not_in=ids_not_in,\n emails_iin=emails_iin,\n emails_in=emails_in,\n emails_not_iin=emails_not_iin,\n emails_not_in=emails_not_in,\n email_ilike=email_ilike,\n email_like=email_like,\n email_not_ilike=email_not_ilike,\n email_not_like=email_not_like,\n email_equal=email_equal,\n email_iequal=email_iequal,\n email_different=email_different,\n email_idifferent=email_idifferent,\n zip_codes_in=zip_codes_in,\n zip_codes_not_in=zip_codes_not_in,\n is_active_equal=is_active_equal,\n load_addresses=load_addresses,\n load_city=load_city,\n order_by=order_by,\n direction=direction,\n )\n\n return [UserRead.model_validate(item) for item in users]\n\n def get_users_paginate(\n self,\n __session__: Session,\n /,\n page: int,\n per_page: int,\n ids_in: Optional[List[int]] = None,\n ids_not_in: Optional[List[int]] = None,\n emails_iin: Optional[List[str]] = None,\n emails_in: Optional[List[str]] = None,\n emails_not_iin: Optional[List[str]] = None,\n emails_not_in: Optional[List[str]] = None,\n email_ilike: Optional[str] = None,\n email_like: Optional[str] = None,\n email_not_ilike: Optional[str] = None,\n email_not_like: Optional[str] = None,\n email_equal: Optional[str] = None,\n email_iequal: Optional[str] = None,\n email_different: Optional[str] = None,\n email_idifferent: Optional[str] = None,\n zip_codes_in: Optional[List[int]] = None,\n zip_codes_not_in: Optional[List[int]] = None,\n is_active_equal: Optional[bool] = None,\n load_addresses: bool = True,\n load_city: bool = True,\n order_by: Optional[List[str]] = None,\n direction: Optional[List[str]] = None,\n ) -> Tuple[List[UserRead], str]:\n users, pagination = self._repository.get_paginate(\n __session__,\n page=page,\n per_page=per_page,\n ids_in=ids_in,\n ids_not_in=ids_not_in,\n emails_iin=emails_iin,\n emails_in=emails_in,\n emails_not_iin=emails_not_iin,\n emails_not_in=emails_not_in,\n email_ilike=email_ilike,\n email_like=email_like,\n email_not_ilike=email_not_ilike,\n email_not_like=email_not_like,\n email_equal=email_equal,\n email_iequal=email_iequal,\n email_different=email_different,\n email_idifferent=email_idifferent,\n zip_codes_in=zip_codes_in,\n zip_codes_not_in=zip_codes_not_in,\n is_active_equal=is_active_equal,\n load_addresses=load_addresses,\n load_city=load_city,\n order_by=order_by,\n direction=direction,\n )\n\n return [UserRead.model_validate(item) for item in users], pagination\n\n def get_user_by_id(\n self,\n __session__: Session,\n /,\n id: int,\n ) -> Optional[UserRead]:\n user = self._repository.get_by_id(\n __session__,\n id=id,\n )\n\n if user is None:\n return None\n\n return UserRead.model_validate(user)\n\n def create_user(\n self,\n __session__: Session,\n /,\n data: UserCreate,\n ) -> UserRead:\n user = self._repository.create(\n __session__,\n data=data,\n flush=True,\n )\n\n return UserRead.model_validate(user)\n\n def create_users(\n self,\n __session__: Session,\n /,\n data: List[UserCreate],\n ) -> List[UserRead]:\n users = self._repository.create_all(\n __session__,\n data=data,\n flush=True,\n )\n\n return [UserRead.model_validate(user) for user in users]\n\n def patch_email(\n self,\n __session__: Session,\n /,\n id: int,\n email: str,\n ) -> UserRead:\n user = self._repository.patch_email(\n __session__,\n id=id,\n email=email,\n flush=True,\n )\n\n return UserRead.model_validate(user)\n\n def bulk_patch_email(\n self,\n __session__: Session,\n /,\n data: List[Tuple[int, str]],\n ) -> List[UserRead]:\n users = self._repository.bulk_patch_email(\n __session__,\n data=data,\n flush=True,\n )\n\n return [UserRead.model_validate(user) for user in users]\n\n def patch_disable(\n self,\n __session__: Session,\n /,\n ids: List[int],\n ) -> List[UserRead]:\n users = self._repository.patch_disable(\n __session__,\n ids=ids,\n flush=True,\n )\n\n return [UserRead.model_validate(user) for user in users]\n\n def delete_by_id(\n self,\n __session__: Session,\n /,\n id: int,\n ) -> bool:\n return self._repository.delete(\n __session__,\n id=id,\n flush=True,\n )\n\n def delete_by_ids(\n self,\n __session__: Session,\n /,\n ids: List[int],\n ) -> bool:\n return self._repository.delete_all(\n __session__,\n ids=ids,\n flush=True,\n )\n```\n\nTo create an instance of AsyncDatabase and generate an instance of AsyncSession from the factory:\n\n```\nimport logging\n\nlogging.basicConfig()\nlogger_db = logging.get_logger('demo')\n\ndatabase = AsyncDataBase(\n databases_config={\n \"connection_string\": \"foo\",\n },\n base=Base,\n logger=logger_db,\n)\n\nawait database.create_database()\n\nasync with database.session_factory() as session:\n await session.execute(...)\n```\n\nTo create an async repository, you just have to inherit your class from AsyncRepository.\n\n```\n# MODULES\nfrom typing import Callable, List, Optional, Tuple\n\n# SQLALCHEMY\nfrom sqlalchemy import Sequence\nfrom sqlalchemy.ext.asyncio import AsyncSession\n\n# PYSQL_REPO\nfrom pysql_repo import Operators\nfrom pysql_repo.asyncio import AsyncRepository\n\n# CONTEXTLIB\nfrom contextlib import AbstractAsyncContextManager\n\n# MODEL\nfrom tests.repositories.user._base import UserRepositoryBase as _UserRepositoryBase\nfrom tests.models.database.database import User\nfrom tests.models.schemas.user import UserCreate\n\n\nclass AsyncUserRepository(AsyncRepository, _UserRepositoryBase):\n def __init__(\n self,\n session_factory: Callable[..., AbstractAsyncContextManager[AsyncSession]],\n ) -> None:\n super().__init__(session_factory)\n\n async def get_all(\n self,\n __session__: AsyncSession,\n /,\n ids_in: Optional[List[int]] = None,\n ids_not_in: Optional[List[int]] = None,\n emails_iin: Optional[List[str]] = None,\n emails_in: Optional[List[str]] = None,\n emails_not_iin: Optional[List[str]] = None,\n emails_not_in: Optional[List[str]] = None,\n email_ilike: Optional[str] = None,\n email_like: Optional[str] = None,\n email_not_ilike: Optional[str] = None,\n email_not_like: Optional[str] = None,\n email_equal: Optional[str] = None,\n email_iequal: Optional[str] = None,\n email_different: Optional[str] = None,\n email_idifferent: Optional[str] = None,\n zip_codes_in: Optional[List[int]] = None,\n zip_codes_not_in: Optional[List[int]] = None,\n is_active_equal: Optional[bool] = None,\n load_addresses: bool = True,\n load_city: bool = True,\n order_by: Optional[List[str]] = None,\n direction: Optional[List[str]] = None,\n ) -> Sequence[User]:\n users = await self._select_all(\n __session__,\n model=User,\n optional_filters=self.get_filters(\n ids_in=ids_in,\n ids_not_in=ids_not_in,\n emails_iin=emails_iin,\n emails_in=emails_in,\n emails_not_iin=emails_not_iin,\n emails_not_in=emails_not_in,\n email_ilike=email_ilike,\n email_like=email_like,\n email_not_ilike=email_not_ilike,\n email_not_like=email_not_like,\n email_equal=email_equal,\n email_iequal=email_iequal,\n email_different=email_different,\n email_idifferent=email_idifferent,\n zip_codes_in=zip_codes_in,\n zip_codes_not_in=zip_codes_not_in,\n is_active_equal=is_active_equal,\n ),\n relationship_options=self.get_relationship_options(\n load_addresses=load_addresses,\n load_city=load_city,\n zip_codes_in=zip_codes_in,\n zip_codes_not_in=zip_codes_not_in,\n ),\n order_by=order_by,\n direction=direction,\n )\n\n return users\n\n async def get_paginate(\n self,\n __session__: AsyncSession,\n /,\n page: int,\n per_page: int,\n ids_in: Optional[List[int]] = None,\n ids_not_in: Optional[List[int]] = None,\n emails_iin: Optional[List[str]] = None,\n emails_in: Optional[List[str]] = None,\n emails_not_iin: Optional[List[str]] = None,\n emails_not_in: Optional[List[str]] = None,\n email_ilike: Optional[str] = None,\n email_like: Optional[str] = None,\n email_not_ilike: Optional[str] = None,\n email_not_like: Optional[str] = None,\n email_equal: Optional[str] = None,\n email_iequal: Optional[str] = None,\n email_different: Optional[str] = None,\n email_idifferent: Optional[str] = None,\n zip_codes_in: Optional[List[int]] = None,\n zip_codes_not_in: Optional[List[int]] = None,\n is_active_equal: Optional[bool] = None,\n load_addresses: bool = True,\n load_city: bool = True,\n order_by: Optional[List[str]] = None,\n direction: Optional[List[str]] = None,\n ) -> Tuple[Sequence[User], str]:\n users, pagination = await self._select_paginate(\n __session__,\n model=User,\n optional_filters=self.get_filters(\n ids_in=ids_in,\n ids_not_in=ids_not_in,\n emails_iin=emails_iin,\n emails_in=emails_in,\n emails_not_iin=emails_not_iin,\n emails_not_in=emails_not_in,\n email_ilike=email_ilike,\n email_like=email_like,\n email_not_ilike=email_not_ilike,\n email_not_like=email_not_like,\n email_equal=email_equal,\n email_iequal=email_iequal,\n email_different=email_different,\n email_idifferent=email_idifferent,\n zip_codes_in=zip_codes_in,\n zip_codes_not_in=zip_codes_not_in,\n is_active_equal=is_active_equal,\n ),\n relationship_options=self.get_relationship_options(\n load_addresses=load_addresses,\n load_city=load_city,\n zip_codes_in=zip_codes_in,\n zip_codes_not_in=zip_codes_not_in,\n ),\n order_by=order_by,\n direction=direction,\n page=page,\n per_page=per_page,\n )\n\n return users, pagination\n\n async def get_by_id(\n self,\n __session__: AsyncSession,\n /,\n id: int,\n ) -> Optional[User]:\n user = await self._select(\n __session__,\n model=User,\n filters={\n User.id: {\n Operators.EQUAL: id,\n },\n },\n )\n\n return user\n\n async def create(\n self,\n __session__: AsyncSession,\n /,\n data: UserCreate,\n flush: bool = False,\n commit: bool = True,\n ) -> User:\n user = await self._add(\n __session__,\n model=User,\n values={\n User.email.key: data.email,\n User.hashed_password.key: data.hashed_password,\n User.full_name.key: data.full_name,\n },\n flush=flush,\n commit=commit,\n )\n\n return user\n\n async def create_all(\n self,\n __session__: AsyncSession,\n /,\n data: List[UserCreate],\n flush: bool = False,\n commit: bool = True,\n ) -> Sequence[User]:\n users = await self._add_all(\n __session__,\n model=User,\n values=[\n {\n User.email.key: item.email,\n User.hashed_password.key: item.hashed_password,\n User.full_name.key: item.full_name,\n }\n for item in data\n ],\n flush=flush,\n commit=commit,\n )\n\n return users\n\n async def patch_email(\n self,\n __session__: AsyncSession,\n /,\n id: int,\n email: str,\n flush: bool = False,\n commit: bool = True,\n ) -> Optional[User]:\n user = await self._update(\n __session__,\n model=User,\n values={\n User.email.key: email,\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 async def bulk_patch_email(\n self,\n __session__: AsyncSession,\n /,\n data: List[Tuple[int, str]],\n flush: bool = False,\n commit: bool = True,\n ) -> Sequence[User]:\n await self._bulk_update(\n __session__,\n model=User,\n values=[\n {\n User.id.key: id,\n User.email.key: email,\n }\n for id, email in data\n ],\n flush=flush,\n commit=commit,\n )\n\n return await self._select_all(\n __session__,\n model=User,\n filters={\n User.id: {\n Operators.IN: [id for id, _ in data],\n },\n },\n )\n\n async def patch_disable(\n self,\n __session__: AsyncSession,\n /,\n ids: List[int],\n flush: bool = False,\n commit: bool = True,\n ) -> Sequence[User]:\n users = await self._update_all(\n __session__,\n model=User,\n values={\n User.is_active.key: False,\n },\n filters={\n User.id: {\n Operators.IN: ids,\n },\n },\n flush=flush,\n commit=commit,\n )\n\n return users\n\n async def delete(\n self,\n __session__: AsyncSession,\n /,\n id: int,\n flush: bool = False,\n commit: bool = True,\n ) -> bool:\n is_deleted = await self._delete(\n __session__,\n model=User,\n filters={\n User.id: {\n Operators.EQUAL: id,\n },\n },\n flush=flush,\n commit=commit,\n )\n\n return is_deleted\n\n async def delete_all(\n self,\n __session__: AsyncSession,\n /,\n ids: List[int],\n flush: bool = False,\n commit: bool = True,\n ) -> bool:\n is_deleted = await self._delete_all(\n __session__,\n model=User,\n filters={\n User.id: {\n Operators.IN: ids,\n },\n },\n flush=flush,\n commit=commit,\n )\n\n return is_deleted\n```\n\nTo create an async service, you just have to inherit your class from AsyncService.\n\n```\n# MODULES\nfrom logging import Logger\nfrom typing import List, Optional, Tuple\n\n# SQLALCHEMY\nfrom sqlalchemy.ext.asyncio import AsyncSession\n\n# PYSQL_REPO\nfrom pysql_repo.asyncio import AsyncService, with_async_session\n\n# REPOSITORIES\nfrom tests.repositories.user.async_user_repository import AsyncUserRepository\n\n# MODELS\nfrom tests.models.schemas.user import UserCreate, UserRead\n\n\nclass AsyncUserService(AsyncService[AsyncUserRepository]):\n def __init__(\n self,\n user_repository: AsyncUserRepository,\n logger: Logger,\n ) -> None:\n super().__init__(\n repository=user_repository,\n )\n self._logger = logger\n\n async def get_users(\n self,\n __session__: AsyncSession,\n /,\n ids_in: Optional[List[int]] = None,\n ids_not_in: Optional[List[int]] = None,\n emails_iin: Optional[List[str]] = None,\n emails_in: Optional[List[str]] = None,\n emails_not_iin: Optional[List[str]] = None,\n emails_not_in: Optional[List[str]] = None,\n email_ilike: Optional[str] = None,\n email_like: Optional[str] = None,\n email_not_ilike: Optional[str] = None,\n email_not_like: Optional[str] = None,\n email_equal: Optional[str] = None,\n email_iequal: Optional[str] = None,\n email_different: Optional[str] = None,\n email_idifferent: Optional[str] = None,\n zip_codes_in: Optional[List[int]] = None,\n zip_codes_not_in: Optional[List[int]] = None,\n is_active_equal: Optional[bool] = None,\n load_addresses: bool = True,\n load_city: bool = True,\n order_by: Optional[List[str]] = None,\n direction: Optional[List[str]] = None,\n ) -> List[UserRead]:\n users = await self._repository.get_all(\n __session__,\n ids_in=ids_in,\n ids_not_in=ids_not_in,\n emails_iin=emails_iin,\n emails_in=emails_in,\n emails_not_iin=emails_not_iin,\n emails_not_in=emails_not_in,\n email_ilike=email_ilike,\n email_like=email_like,\n email_not_ilike=email_not_ilike,\n email_not_like=email_not_like,\n email_equal=email_equal,\n email_iequal=email_iequal,\n email_different=email_different,\n email_idifferent=email_idifferent,\n zip_codes_in=zip_codes_in,\n zip_codes_not_in=zip_codes_not_in,\n is_active_equal=is_active_equal,\n load_addresses=load_addresses,\n load_city=load_city,\n order_by=order_by,\n direction=direction,\n )\n\n return [UserRead.model_validate(item) for item in users]\n\n async def get_users_paginate(\n self,\n __session__: AsyncSession,\n /,\n page: int,\n per_page: int,\n ids_in: Optional[List[int]] = None,\n ids_not_in: Optional[List[int]] = None,\n emails_iin: Optional[List[str]] = None,\n emails_in: Optional[List[str]] = None,\n emails_not_iin: Optional[List[str]] = None,\n emails_not_in: Optional[List[str]] = None,\n email_ilike: Optional[str] = None,\n email_like: Optional[str] = None,\n email_not_ilike: Optional[str] = None,\n email_not_like: Optional[str] = None,\n email_equal: Optional[str] = None,\n email_iequal: Optional[str] = None,\n email_different: Optional[str] = None,\n email_idifferent: Optional[str] = None,\n zip_codes_in: Optional[List[int]] = None,\n zip_codes_not_in: Optional[List[int]] = None,\n is_active_equal: Optional[bool] = None,\n load_addresses: bool = True,\n load_city: bool = True,\n order_by: Optional[List[str]] = None,\n direction: Optional[List[str]] = None,\n ) -> Tuple[List[UserRead], str]:\n users, pagination = await self._repository.get_paginate(\n __session__,\n page=page,\n per_page=per_page,\n ids_in=ids_in,\n ids_not_in=ids_not_in,\n emails_iin=emails_iin,\n emails_in=emails_in,\n emails_not_iin=emails_not_iin,\n emails_not_in=emails_not_in,\n email_ilike=email_ilike,\n email_like=email_like,\n email_not_ilike=email_not_ilike,\n email_not_like=email_not_like,\n email_equal=email_equal,\n email_iequal=email_iequal,\n email_different=email_different,\n email_idifferent=email_idifferent,\n zip_codes_in=zip_codes_in,\n zip_codes_not_in=zip_codes_not_in,\n is_active_equal=is_active_equal,\n load_addresses=load_addresses,\n load_city=load_city,\n order_by=order_by,\n direction=direction,\n )\n\n return [UserRead.model_validate(item) for item in users], pagination\n\n async def get_user_by_id(\n self,\n __session__: AsyncSession,\n /,\n id: int,\n ) -> Optional[UserRead]:\n user = await self._repository.get_by_id(\n __session__,\n id=id,\n )\n\n if user is None:\n return None\n\n return UserRead.model_validate(user)\n\n async def create_user(\n self,\n __session__: AsyncSession,\n /,\n data: UserCreate,\n ) -> UserRead:\n user = await self._repository.create(\n __session__,\n data=data,\n flush=True,\n )\n\n return UserRead.model_validate(user)\n\n async def create_users(\n self,\n __session__: AsyncSession,\n /,\n data: List[UserCreate],\n ) -> List[UserRead]:\n users = await self._repository.create_all(\n __session__,\n data=data,\n flush=True,\n )\n\n return [UserRead.model_validate(user) for user in users]\n\n async def patch_email(\n self,\n __session__: AsyncSession,\n /,\n id: int,\n email: str,\n ) -> UserRead:\n user = await self._repository.patch_email(\n __session__,\n id=id,\n email=email,\n flush=True,\n )\n\n return UserRead.model_validate(user)\n\n async def bulk_patch_email(\n self,\n __session__: AsyncSession,\n /,\n data: List[Tuple[int, str]],\n ) -> List[UserRead]:\n users = await self._repository.bulk_patch_email(\n __session__,\n data=data,\n flush=True,\n )\n\n return [UserRead.model_validate(user) for user in users]\n\n async def patch_disable(\n self,\n __session__: AsyncSession,\n /,\n ids: List[int],\n ) -> List[UserRead]:\n users = await self._repository.patch_disable(\n __session__,\n ids=ids,\n flush=True,\n )\n\n return [UserRead.model_validate(user) for user in users]\n\n async def delete_by_id(\n self,\n __session__: AsyncSession,\n /,\n id: int,\n ) -> bool:\n return await self._repository.delete(\n __session__,\n id=id,\n flush=True,\n )\n\n async def delete_by_ids(\n self,\n __session__: AsyncSession,\n /,\n ids: List[int],\n ) -> bool:\n return await self._repository.delete_all(\n __session__,\n ids=ids,\n flush=True,\n )\n\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.7.4.0",
"project_urls": {
"Download": "https://github.com/Impro02/pysql-repo/archive/refs/tags/0.7.4.0.tar.gz",
"Homepage": "https://github.com/Impro02/pysql-repo"
},
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "54c1318eca0019bb831a75948c7881adea9b2c28f1e57aaecfbbfbd7e8e7aabf",
"md5": "0036db14248d3d4ed51696ad5f83fab5",
"sha256": "e54e86fd34df1e44c7a0ad0c7106034b9e731606d4b7fb93656f790bb5f9ecba"
},
"downloads": -1,
"filename": "pysql_repo-0.7.4.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "0036db14248d3d4ed51696ad5f83fab5",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 27053,
"upload_time": "2024-07-16T09:38:59",
"upload_time_iso_8601": "2024-07-16T09:38:59.080799Z",
"url": "https://files.pythonhosted.org/packages/54/c1/318eca0019bb831a75948c7881adea9b2c28f1e57aaecfbbfbd7e8e7aabf/pysql_repo-0.7.4.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "4819bacf94b17a2352cf8dd736da1b6e6ff1170aa584787031380bf816a8ca28",
"md5": "016bc660898e16b93728103fcd806857",
"sha256": "4e9a59780d67eddb4d61bf077d7faba8b5e29d0c6316a0839c9b3e42b6a2c526"
},
"downloads": -1,
"filename": "pysql-repo-0.7.4.0.tar.gz",
"has_sig": false,
"md5_digest": "016bc660898e16b93728103fcd806857",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 28245,
"upload_time": "2024-07-16T09:39:00",
"upload_time_iso_8601": "2024-07-16T09:39:00.885115Z",
"url": "https://files.pythonhosted.org/packages/48/19/bacf94b17a2352cf8dd736da1b6e6ff1170aa584787031380bf816a8ca28/pysql-repo-0.7.4.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-07-16 09:39:00",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "Impro02",
"github_project": "pysql-repo",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"requirements": [],
"lcname": "pysql-repo"
}
