No project description provided
Project description
easy-crud-repo-service
Package that allows user to use any object as entity and process basic CRUD operations on database.
Installation
Using pip:
pip install easy-crud-repo-service
Using poetry:
poetry add easy-crud-repo-service
Using pipenv:
pipenv install easy-crud-repo-service
TESTS
1. Clone repository and enter main directory:
git clone https://github.com/DSmolke/easy_crud_repo_service
cd easy_crud_repo_service
1. Make sure your docker is running and port 3306 is empty, and run docker-compose command
docker-compose up -d
cd tests
3. Run tests using:
####Poetry:
poetry install
poetry shell
cd tests
poetry run pytest -vv
####Pipenv:
pipenv shell
cd tests
pipenv run pytest -vv
####Pip:
pip install mysql-connector-python
pip install python-dotenv
pip install easyvalid-data-validator
pip install inflection
pip install pytest
pip install dbm-database-service
pip install coverage
cd tests
pytest -vv
Basic usage
Having existing database server, or mysql container, we want to make some
###1. Step - prepare .env file Please download .env file template - > link
Edit file according to your needs and place it in same directory.
POOL_NAME=TEST
POOL_SIZE=5
POOL_RESET_SESSION=True
HOST=localhost
DATABASE=db_1
USER=root
PASSWORD=root
PORT=3306
###2. Step - Import all necessary objects
from pathlib import Path
from easy_crud_repo_service.repo.crud_repo import CrudRepo
from easy_crud_repo_service.repo.connections.builders import MySQLConnectionPoolBuilder
###3. Step - Create new MySQLConnectionPoolBuilder
dbm_builder = MySQLConnectionPoolBuilder(f"{Path.cwd()}\\.env")
###4. Step - Create new MySQLDatabaseManager using object prepared in previous step
dbm = dbm_builder.build()
###5. Create crud repo according to your needs
# Entity that you want to work on
@dataclass
class Team:
id: int = 0
name: str = ""
points: int = 0
crud_repo = CrudRepo(dbm, Team)
###5. Make operations on database table
crud_repo.insert(Team(1, 'REAL MADRIT', 20))
print(crud_repo.find_all())
Objects
Connectors
There are two types of connectors available. get_connection_pool and MySQLConnectionPoolBuilder. They differ in philosophy, but work on the same principle. To load environmental variables needed for db connection and return connection pool.
get_connection_pool
Example:
get_connection_pool(<ABSOLUTE-PATH>) -> full path of .env file can be provided
MySQLConnectionPoolBuilder
Example:
# The principle for .env path is same
builder = MySQLConnectionPoolBuilder(<ABSOLUTE-PATH>)
# you can modify credentials during program flow
builder.set_new_port(3307).add_new_password('password')
# when you are ready, you can build connection pool
database_manager = builder.build()
CrudRepo
It takes class as an entity and uses its properties to prepare sql statements for communication with database
class CrudRepo:
def __init__(self, connection_pool: pooling.MySQLConnectionPool, entity: type) -> None:
self._connection_pool = connection_pool
self._entity = entity
self._entity_type = type(entity())
Methods:
_get_cursor_object:
Example:
@contextmanager
def _get_cursor_object(self):
connection_object = self._connection_pool.get_connection()
try:
if connection_object.is_connected():
cursor_object = connection_object.cursor()
yield cursor_object
connection_object.commit()
except Error as e:
connection_object.rollback()
finally:
if connection_object.is_connected():
cursor_object.close()
connection_object.close()
Context manager that allows us to work on 'with' statement to avoid problems when errors occur during operations on cursor
_table_name:
Example:
def _table_name(self) -> str:
return inflection.tableize(self._entity_type.__name__)
Creates valid MySQL table name by adding s at the end of the class name as well as lowering letters case Team -> teams
_fields_names:
Example:
def _fields_names(self) -> list[str]:
return list(self._entity().__dict__.keys())
Method returns names of objects attributes as a list
_column_names_for_insert:
Example:
def _column_names_for_insert(self) -> str:
return ', '.join([field for field in self._fields_names() if field.lower() != 'id'])
Returns all names accept id
insert:
Example:
def insert(self, item: Any) -> int:
with self._get_cursor_object() as cur:
sql = f'insert into {self._table_name()} ' \
f'({self._column_names_for_insert()}) ' \
f'values ({self._column_values_for_insert(item)});'
cur.execute(sql)
return cur.lastrowid
Method inserts object into table and return its id
insert_many:
Example:
def insert_many(self, items: list[Any]) -> list[int]:
with self._get_cursor_object() as cur:
values = ", ".join([f"({CrudRepo._column_values_for_insert(item)})" for item in items])
sql = f"insert into {self._table_name()} ({self._column_names_for_insert()}) values {values}"
cur.execute(sql)
return [item.id for item in self.find_n_last(len(items))]
Method inserts objects into table and return theirs ids
update:
Example:
def update(self, item_id: int, item: Any) -> Any:
with self._get_cursor_object() as cur:
sql = f"update {self._table_name()} set {CrudRepo._column_names_and_values_for_update(item)} where id = {item_id}"
cur.execute(sql)
return self.find_one(item_id)
Method updates row that has specified id with provided object
find_n_last:
Example:
def find_n_last(self, n) -> list[Any]:
with self._get_cursor_object() as cur:
sql = f'select * from {self._table_name()} order by id desc limit {n}'
cur.execute(sql)
return [self._entity(*row) for row in cur.fetchall()]
Method finds n last rows of table that we are working on
find_one:
Example:
def find_one(self, item_id: int) -> Any:
with self._get_cursor_object() as cur:
sql = f"select * from {self._table_name()} where id = {item_id}"
cur.execute(sql)
result = cur.fetchone()
if not result:
raise RuntimeError(f"Item with id {item_id} wasn't found")
return self._entity(*result)
Method finds row by id
find_all:
Example:
def find_all(self) -> list[Any]:
with self._get_cursor_object() as cur:
sql = f"select * from {self._table_name()}"
cur.execute(sql)
return [self._entity(*row) for row in cur.fetchall()]
Method finds all rows in the table
delete_one:
Example:
def delete_one(self, item_id) -> int:
with self._get_cursor_object() as cur:
sql = f"delete from {self._table_name()} where id = {item_id}"
self.find_one(item_id)
cur.execute(sql)
return item_id
Method deletes one row using provided id
delete_all:
Example:
def delete_all(self) -> list[int]:
with self._get_cursor_object() as cur:
all_deleted_items = [item.id for item in self.find_all()]
sql = f'delete from {self._table_name()} where id >= 1'
cur.execute(sql)
return all_deleted_items
Method deletes all rows from table
_column_values_for_insert:
Example:
@classmethod
def _column_values_for_insert(cls, item: Any) -> str:
def to_str(entry: Any) -> str:
return f"'{entry[1]}'" if isinstance(entry[1], (str, datetime, date)) else str(entry[1])
return ", ".join([to_str(entry) for entry in item.__dict__.items() if entry[0].lower() != 'id'])
Method creates expression with values that we want to put into sql. (All accept id)
_column_names_and_values_for_update:
Example:
@classmethod
def _column_names_and_values_for_update(cls, entity) -> str:
def to_str(entry: Any) -> str:
return entry[0] + '=' + (f"'{entry[1]}'" if isinstance(entry[1], (str, datetime, date)) else str(entry[1]))
return ', '.join([to_str(item) for item in entity.__dict__.items() if item[0].lower() != 'id'])
Method creates expression that will be responsible for updating row values. (All accept id)
Project details
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for easy_crud_repo_service-0.1.6.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | c12e0a74b877955b13aee3e9521c731a6c0d36b2c2c1b26ac2fabec49eea9cec |
|
MD5 | 3582a5ba467662eb66eb9701a76369b6 |
|
BLAKE2b-256 | 4bb86df7963c3e610983010600ac65b9d1563a0694cb10ac4ad7be0a3370f875 |
Hashes for easy_crud_repo_service-0.1.6-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 6f6ab3cb2f924ed854c303d849209c9e514b15eab42e16af48a06aa298cb0ce8 |
|
MD5 | 6173d4a00d8cd7452d95681d6bb828a7 |
|
BLAKE2b-256 | 7176ccbdf73652f2a8ec6d6693ea4b0d9baad82512fbc5b944165fda59c87c71 |