| Name | alphonse JSON |
| Version |
0.1.0
JSON |
| download |
| home_page | |
| Summary | A wrapper around SQLAlchemy to make it easier to make more complex queries |
| upload_time | 2024-02-13 19:48:45 |
| maintainer | Jackson Prince |
| docs_url | None |
| author | Jackson Prince |
| requires_python | >=3.8 |
| license | Copyright (c) 2024 Wavelynx Technologies Corporation Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. |
| keywords |
python
sqlalchemy
|
| VCS |
|
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# Alphonse Database Manager
## What is?
Alphonse is an interface that makes it easier to manage databases using SQLAlchemy.
It provides a set of methods that make complex database queries more performant and easier to write.
For a project already leveraging SQLAlchemy, it only requires a few lines of code to start using it immediately.
---
## Getting started
### Requirements
- SQLAlchemy version `1.4` or higher (`2.0+` for best performance)
- A SQLAlchemy "Engine" object.
- This is the connection to the database. It can be created using the `sqlalchemy.create_engine` method.
- A list of all defined SQLAlchemy models that all inherit from the same instance of a `sqlalchemy.ext.declarative.declarative_base.`
### Example setup
```python
""" Example setup for Alphonse Database Manager """
from alphonse import DbManager
from sqlalchemy import create_engine
from sqlalchemy.engine import Engine
from my_models import MyModel1, MyModel2, MyModel3
# Create the singelton instance of the engine
db_connection_url: str = "postgresql://user:password@localhost:port/my_db_name"
engine: Engine = create_engine(db_connection_url)
# Create a list of all defined models that all inherit from the same
# instance of a "declarative_base" from sqlalchemy.ext.declarative
model_list: list = [MyModel1, MyModel2, MyModel3]
# Initialize the db_manager
db_manager: DbManager = DbManager(engine, model_list)
```
---
## **Methods and usage**
The main interface to making queries is the singleton `DbManager` object created in the setup.
It provides a set of methods that make database queries more performant and easier to write.
This singleton instance acts as an abstraction layer to prevent circular imports and to provide
a single point of access to the database, ensuring that all sessions are managed correctly.
---
### **`create()`**
The `db_manager.create()` method is used to create new rows in the database.
A boolean is returned to determine if the creation was successful.
It takes two arguments:
- `table_key`: A string representing the model name. This will be the same as the class name of the model.
- `req_payload`: A dictionary mapping of the columns and their values representing the row to be created.
##### **Example:**
```python
from db.objects import db_manager
# Create a new row in the "User" table
creation_map: dict = {
"username": "new_user",
"password": "password123",
"email": "test@testing.net"
}
creation_was_successful: bool = db_manager.create("User", creation_map)
# creation_was_successful = True or False depending on if the creation was successful.
```
#### `map()` method
If your table model has specific requirements for creating a new row, you can define a
`map()` method directly on the model.
This method must take in a dictionary mapping of the columns, assign the key value pairs to the
attributes of the model, and return a mapped instance of the model class or `None` if the mapping fails.
The returned class instance must be a valid instance of the model with all required fields populated
(excepting nullable fields and auto-incrementing fields like primary key ids).
The `map()` method will be called automatically when the `db_manger.create()` method is used
and does not need to be called directly.
This means that the `req_payload` dictionary passed to
the `create()` method only needs to include the values to be assigned to the model in the user-defined `map()` method.
##### **Example model:**
```python
""" User table model. """
from typing import Optional
from sqlalchemy import Column, Integer, String, VARCHAR
from sqlalchemy.orm import relationship
# The below imports are theorhetical and are not included in the package.
# --------------------------------------------------------------------
# Base is the instatiated singleton instance of `declarative_base` orignally imported from
# sqlalchemy.ext.declarative. All models in this theorhetical project should inherit from this same instance.
from db.objects.base import Base
from utils.logger import logger
class User(Base):
""" User orm model. """
__tablename__ = "users"
id = Column(Integer, primary_key=True, nullable=False)
name = Column(String(50), unique=True, nullable=False)
status = Column(String(20), nullable=False)
website = Column(VARCHAR(255), nullable=True)
@classmethod
def map(cls, req_payload: dict) -> Optional["User"]:
"""
Map the `request_payload` dictionary to the User model.
:param req_payload: A dictionary mapping of the columns in the table and their values representing the new row to be created.
:return: A mapped instance of the User model or None if the mapping fails.
"""
try:
# Create a new instance of the User model.
new_user: User = User(
# Use the `.get()` method to safely access the dictionary values and return None as a default if the key is not found.
name=req_payload.get("name"),
status=req_payload.get("status"),
partner_key=req_payload.get("partner_key"),
# In this example, the id column is an auto-incrementing primary key and the website column is nullable.
# Therefore, they not required to be filled in when the row is created. If this method wasn't defined,
# the `db_manager.create()` method would fail if the `req_payload` dictionary didn't include a `website` value.
)
# Return the mapped instance of the User model.
return new_users
except (KeyError, AttributeError):
logger.log_error("Error mapping the request payload to the User model.")
# If the mapping fails, None is the expected return value to indicate that the mapping failed.
return None
```
---
### **`read()`**
The read method is used to retrieve rows from a single table in the database.
A dictionary of result(s) is returned, or None if the query fails.
It takes two required arguments and one optional argument:
- `table_key`: A string representing the model name. This will be the same as the class name of the model.
- `search_params`: A dictionary mapping of the columns in the table and their values.
Represents the search criteria for the query.
See the `search_params argument` section below for more information.
- `select_params` An optional list of strings representing the columns to select from the db.
Represents the filter parameters for the query.
See the `select_params argument` section below for more information.
##### -**`search_params` argument**
The values in the `search_params` dictionary are used to filter the results of the query.
You can supply a single value or a list of values for each key in this dictionary (column in the table.)
For example: `{"status": "ACTIVE"}` or `{"status": ["ACTIVE", "SUSPENDED"]}`
Select the rows where the status is "ACTIVE" vs. select the rows where the status is "ACTIVE" or "SUSPENDED."
##### -**`select_params` argument**
The `select_params` argument is an optional list of strings representing the columns to select from the db.
They can be used to return partial data from the table if only certain values are needed.
The strings must match the column names in the table exactly as they are defined on the models.
If a valid list is provided, only the columns in the list will be returned.
Returns the full table row if the list is empty or not provided.
For example: `["id", "status"]`
Whatever the `search_params` return, only return the 'id' and 'status'
columns (and their value) for any results from the queried table.
#### **Example queries:**
```python
from typing import Optional
from db.objects import db_manager
# =============
# BASIC EXAMPLE
# =============
# Return all rows from the "User" table with the a status of "ACTIVE".
user: Optional[dict] = db_manager.read(
"User", {"status": "ACTIVE"}
)
# If one result is found:
# user = {
# "id": 1,
# "name": "test_user",
# "status": "ACTIVE",
# "website": "www.testwebsite.com"
# }
# If multiple results are found, they will be within a list at a key of "result."
# user = {
# "result": [
# {"id": 1, "name": "test_user", "status": "ACTIVE", "website": "www.testwebsite.com"},
# {"id": 55, "name": "test_user_55", "status": "ACTIVE", "website": None}
# ]
# }
# If the no rows are found meeting the criteria:
# user = {}
# If an exception was raised during the read operation
# user = None
# ====================================
# EXAMPLE USING MULTIPLE SEARCH PARAMS
# ====================================
# Return all rows from the "User" table with the a status of "ACTIVE" or "SUSPENDED".
user: Optional[dict] = db_manager.read(
"User", {"status": ["ACTIVE", "SUSPENDED"]}
)
# If multiple results are found,
# user = {
# "result": [
# {"id": 1, "name": "test_user", "status": "ACTIVE", "website": "www.testwebsite.com"},
# {"id": 55, "name": "test_user_55", "status": "ACTIVE", "website": None},
# {"id": 55, "name": "test_user_56", "status": "SUSPENDED", "website": "www.othertestwebsite.com"}
# ]
# }
# ===========================
# EXAMPLE USING SELECT PARAMS
# ===========================
# Return the id and status of all active users.
user: Optional[dict] = db_manager.read(
"User",
{"status": "ACTIVE"},
["id", "status",]
)
# If one result is found:
# user = {
# "id": 1,
# "status": "ACTIVE",
# }
```
---
### **`update()`**
The update method is used to edit existing rows in the database.
A boolean is returned to determine if the creation was successful.
It takes three arguments:
- `table_key`: A string representing the model name. This will be the same as the class name of the model.
- `search_params`: A dictionary mapping of parameters pertinent to specifying the query. Represents the search criteria for the query. All columns in the dictionary must be present in the table. See the `search_params argument` section below for more information.
- `insert_params`: Mapped dictionary of key/value pairs corresponding to db columns to be updated.
All columns in the dictionary must be present in the table.
Operations that leave orphaned rows will not be performed and will result in the operation failing.
##### -**`search_params` argument**
The values in the `search_params` dictionary are used to filter the results of the query.
You can supply a single value or a list of values for each key in this dictionary (column in the table.)
For example: `{"status": "ACTIVE"}` or `{"status": ["ACTIVE", "SUSPENDED"]}`
Select the rows where the status is "ACTIVE" vs. select the rows where the status is "ACTIVE" or "SUSPENDED."
#### **Example queries:**
```python
from db.objects import db_manager
# Find the row in the "User" table with the id of 1 and update the website column.
params_to_update: dict = {"website": "www.newwebsite.com"}
update_was_successful: bool = db_manager.update(
"User",
{"id": 1},
params_to_update
)
# update_was_successful = True or False depending on if the update was successful
# Find the all rows in the "User" table with a status of "ACTIVE" or "SUSPENDED" and update the status column to "DELETED"
update_was_successful: bool = db_manager.update(
"User",
{"status": ["ACTIVE", "SUSPENDED"]},
{"status": "DELETED"}
)
# update_was_successful = True or False depending on if the update was successful
```
---
### **`delete()`**
The delete method is used to remove existing rows from the database.
A boolean is returned to determine if the creation was successful.
It takes two arguments:
- `table_key`: A string representing the model name. This will be the same as the class name of the model.
- `search_params`:A dictionary mapping of parameters pertinent to specifying the query. Represents the search criteria for the query. All columns in the dictionary must be present in the table. See the `search_params argument` section below for more information.
##### -**`search_params` argument**
The values in the `search_params` dictionary are used to filter the results of the query.
You can supply a single value or a list of values for each key in this dictionary (column in the table.)
For example: `{"status": "ACTIVE"}` or `{"status": ["ACTIVE", "SUSPENDED"]}`
Select the rows where the status is "ACTIVE" vs. select the rows where the status is "ACTIVE" or "SUSPENDED."
#### **Example queries:**
```python
from db.objects import db_manager
# Find the row in the "User" table with the id of 1 and delete it.
delete_was_successful: bool = db_manager.delete("User", {"id": 1})
# delete_was_successful = True or False depending on if the delete was successful
# Find all row(s) in the "User" table with a status of "ACTIVE" or "SUSPENDED" and delete them.
delete_was_successful: bool = db_manager.delete("User", {"status": ["DELETE", "SUSPENDED"]})
# delete_was_successful = True or False depending on if the delete was successful
```
---
### **`joined_read()`**
The joined_read method is used to retrieve rows from multiple tables in the database.
A dictionary of results is returned, or None is returned if the query fails.
It takes three required arguments and one optional argument:
- `starting_table`: A string representing the table where the read should start looking. This will be the same as the class name of the model.
- `ending_table`: A string representing the table where the read should, inclusively, stop looking. This will be the same as the class name of the model.
- `search_params`: This can be one of two datastructures:
- A dictionary mapping of the columns in the starting table and their values representing the search criteria for the query.
- A list of dictionary mappings each representing a table that will be traversed.
Represents the search criteria for each table (in order traversed) for the query.
See `search_params argument` section below for more information.
- `select_params` An optional list representing the columns to select from the db to be used as filter parameters for the query.
This can be one of two datastructures:
- A list of strings representing the columns to select from the starting table.
- A list of lists containing strings representing the columns to select from each table in the order they are traversed.
See `select_params argument` section below for more information.
#### -**search_params argument**
- If only a single dict of `search_prams` is provided,
the JOINS statement will find all rows from related tables with a foreign key pointing at the found of the starting table.
For example, if the `starting_table` is the "User" table, the list of `search_params` could look like:
```python
# In these examples there are three related tables: "User", "Post", "Comments" and "Likes".
# A User can have many Posts, a Post can have many Comments and a comment can have many Likes.
db_manager.joined_read(
"User", "Comments", {"id": 1}
)
# Or
db_manager.joined_read(
"User", "Comments", [{"id": 1}]
)
# This reads as:
# find the User with an 'id' of 1,
# then find the all Posts that have a 'user_id' of 1,
# then find all Comments that have a 'post_id' that matches any of the found Posts.
```
You can also use a list of values to broaden the search criteria, just like in the `read()` method.
```python
db_manager.joined_read(
"User", "Comments", {"status": ["ACTIVE", "SUSPENDED"]}
)
# Or
db_manager.joined_read(
"User", "Comments", [{"status": ["ACTIVE", "SUSPENDED"]}]
)
# This reads as:
# find the all Users with a status of "ACTIVE" or "SUSPENDED",
# then find the all Posts that have a 'user_id's that match any of the found Users,
# then find all Comments that have a 'post_id' that matches any of the found Posts.
```
- If a list of these dictionaries is supplied, it must be the same length as the number of tables to be traversed
in the order that they are traversed.
An empty dict is supplied if no additional search criteria is needed for a table in the JOINS statement.
For example, if the starting table is "User"
from the below examples and the ending table is "Likes," the list of `search_param` would look like:
```python
# In these examples there are three related tables: "User", "Post", "Comments" and "Likes".
# A User can have many Posts, a Post can have many Comments and a comment can have many Likes.
db_manager.joined_read(
"User", "Likes", [{"id": 1}, {"title": "test_post"}, {}]
)
# This reads as find the User with an 'id' of 1,
# then find the Post with a 'user_id' of 1 and a 'title' of "test_post,"
# then find all Likes that have a 'post_id' that matches the id(s) of the Post called "test_post."
```
#### -**select_params argument**
- If no `select_params` are provided, the full row of each table will be returned.
- If only a single list of `select_params` is provided,
the JOINS statement will only apply the filter to the first table in the JOINS statement.
For example, if the `starting_table` is the "User" table from the below examples and
a filter is applied, the list of select params would look like:
- `["name"],` or `[["name"]]`
This reads as, "whatever the `search_params` find,
only return the 'name' column for any results from the User table."
- If a list of these lists is supplied, the filter is applied in order as the tables are traversed. For example:
- `[["name"],[],["id", "content"]]`
This reads as, "whatever the `search_params` find, only return
the 'name' column for any results from the User table,
the all columns (or the full row) for any results from the Post table,
and only return the 'id' and 'content' columns for any results from the Comments table."
#### **Example queries:**
```python
from typing import Optional
from db.objects import db_manager
# In these examples there are three related tables: "User", "Post", "Comments" and "Likes".
# A User can have many Posts, a Post can have many Comments and a comment can have many Likes.
# =============
# BASIC EXAMPLE
# =============
# Return the user with an 'id' of 1, all of the user's posts, & all post's comments.
result_object: Optional[dict] = db_manager.joined_read(
"User",
"Comments",
{"id": 1}
)
# If some results are found:
# result_object: dict = {
# "User": [
# {"id": 1, "name": "test_user", "status": "ACTIVE", "website": "www.testwebsite.com"}
# ],
# "Posts": [
# {"id": 1, "user_id": 1, "title": "test_post", "content": "This is a test post."},
# {"id": 2, "user_id": 1, "title": "test_post_2", "content": "This is a test post."}
# ],
# "Comments": [
# {"id": 1, "post_id": 1, "content": "This is a test comment."},
# {"id": 2, "post_id": 1, "content": "This is a test comment."},
# {"id": 3, "post_id": 2, "content": "This is a test comment."},
# ]
# }
# If no results are found:
# result_object: dict = {}
# If an exception was raised during the read operation
# result_object: dict = None
# ===========================
# EXAMPLE USING SELECT PARAMS
# ===========================
# Return the name of the user with an 'id' of 1, all of the user's posts, & all posts' comments.
result_object: Optional[dict] = db_manager.joined_read(
"User",
"Comments",
{"id": 1},
["name"]
)
# If some results are found:
# result_object: dict = {
# "User": [{"name": "test_user"}],
# "Posts": [
# {"id": 1, "user_id": 1, "title": "test_post", "content": "This is a test post."},
# {"id": 2, "user_id": 1, "title": "test_post_2", "content": "This is a test post."}
# ],
# "Comments": [
# {"id": 1, "post_id": 1, "content": "This is a test comment."},
# {"id": 2, "post_id": 1, "content": "This is a test comment."},
# {"id": 3, "post_id": 2, "content": "This is a test comment."},
# ]
# }
# ====================================
# EXAMPLE USING MULTIPLE SEARCH PARAMS
# ====================================
# Return the the user with the id of 1, the post belonging to the user with a title of "test_post", & all comments belonging to the post.
result_object: Optional[dict] = db_manager.joined_read(
"User",
"Comments",
[{"id": 1}, {"title": "test_post"}, {}],
)
# If some results are found:
# result_object: dict = {
# "User": [
# {"id": 1, "name": "test_user", "status": "ACTIVE", "website": "www.testwebsite.com"}
# ],
# "Posts": [
# {"id": 1, "user_id": 1, "title": "test_post", "content": "This is a test post."},
# ],
# "Comments": [
# {"id": 1, "post_id": 1, "content": "This is a test comment."},
# {"id": 2, "post_id": 1, "content": "This is a test comment."},
# ]
# }
# ====================================
# EXAMPLE USING MULTIPLE SELECT PARAMS
# ====================================
# Return the name of the user with an "id" of 1, full rows ofall of that user's posts, & the "id" and "content" of all posts'"comments."
result_object: Optional[dict] = db_manager.joined_read(
"User",
"Comments",
{"id": 1},
[["name"],[],["id", "content"]]
)
# If some results are found:
# result_object: dict = {
# "User": [
# {"name": "test_user"}
# ],
# "Posts": [
# {"id": 1, "user_id": 1, "title": "test_post", "content": "This is a test post."},
# {"id": 2, "user_id": 1, "title": "test_post_2", "content": "This is a test post."}
# ],
# "Comments": [
# {"id": 1, "content": "This is a test comment."},
# {"id": 2, "content": "This is a test comment."},
# {"id": 3, "content": "This is a test comment."},
# ]
# }
# ===============================================================
# EXAMPLE USING MULTIPLE SEARCH PARAMS AND MULTIPLE SELECT PARAMS
# ===============================================================
# Return the 'name' of the user with an `id` of 1, all posts belonging to the user with a title of "test_post", & and the 'id' and 'conent' of each comment belonging to the post.
result_object: Optional[dict] = db_manager.joined_read(
"User",
"Comments",
[{"id": 1}, {"title": "test_post"}, {}],
[["name"],[],["id", "content"]]
)
# If some results are found:
# result_object: dict = {
# "User": [
# {"name": "test_user"}
# ],
# "Posts": [
# {"id": 1, "user_id": 1, "title": "test_post", "content": "This is a test post."},
# ],
# "Comments": [
# {"id": 1, "content": "This is a test comment."},
# {"id": 2, "content": "This is a test comment."},
# {"id": 3, "content": "This is a test comment."},
# ]
# }
```
---
### **`count()`**
The count method is used to count existing rows that meet criteria in the database.
A dictionary is returned with a count of the rows that meet the criteria, or None is returned if the count fails.
It takes two arguments:
- `table_key`: A string representing the model name. This will be the same as the class name of the model.
- `search_params`:A dictionary mapping of parameters pertinent to specifying the query. Represents the search criteria for the query. All columns in the dictionary must be present in the table. See the `search_params argument` section below for more information.
##### -**`search_params` argument**
The values in the `search_params` dictionary are used to filter the results of the query.
You can supply a single value or a list of values for each key in this dictionary (column in the table.)
#### **Example queries:**
```python
from db.objects import db_manager
# Count the number of rows in the "User" table that have a status of "DELETED".
count: dict = db_manager.count("User", {"status": "DELETED"})
# Count the number of rows in the "User" table that have a status of "DELETED" or "SUSPENDED".
count: dict = db_manager.count("User", {"status": ["DELETED", "SUSPENDED"]})
# If no rows are found meeting the search criteria:
# count = {"count": 0}
# If some rows are found meeting the search criteria:
# count = {"count": 5}
# If an exception was raised during the count operation:
# count = None
```
---
### **Advanced Options**
Certain methods have additional options that can be used to further specify the query.
#### **search_params options**
- Available to the `read()`, `update()`, `delete()`, `joined_read()`, and `count()` methods.
- equality operators:
- You can apply equality operators to the search parameters concatenating the operator to the end of the column name key. The valid operators are:
- `==` for "equals" (default if no operator is provided)
- `!=` for "not equals"
- `<` for "less than"
- `<=` for "less than or equal to"
- `>` for "greater than"
- `>=` for "greater than or equal to"
For example, if you want to return all rows from the "User"
table where the "id" is greater than 5, you would use: `{"id>": 5}`.
- If a column name has no operator concatenated to the end of it, the operator will be used instead of the default "==" operator.
- The data type of the value must be compatible with the operator used. For example, if you use the "<" operator, the value must be a number or a date. If the operator is not valid for the column type, the query will fail.
- If an equality operator is used when multiple values are provided for a column,
the operator will be applied to each value in the list.
For example, if you use `{"status!=": ["ACTIVE", "SUSPENDED"]}`,
the query will return all rows where the status is not "ACTIVE"
or "SUSPENDED."
#### **Example queries:**
```python
import datetime
from db.objects import db_manager
# Return all rows from the "User" table with the a status is not "ACTIVE" or "SUSPENDED".
users: dict = db_manager.read(
"User", {"status!=": ["ACTIVE", "SUSPENDED"]}
)
# Delete all rows from the Users table with an "id" that is less than or equal to 4000 and a status of "DELETED."
delete_was_succesful: bool = db_manager.delete(
"User", {"id<=": 4000, "status": "DELETED"}
)
# Find all rows in the User table that were created on or before October 1, 2021
# with a "status" of "DELETED" or "SUSPENDED" and update each rows' "status" to "ACTIVE".
update_was_successful: bool = db_manager.update(
"User",
{"created_date<=": datetime.date(year=2021, month=10, day=1), "status": ["DELETED", "SUSPENDED"]},
{"status": "ACTIVE"}
)
```
#### **select_params options**
- Available to the `read()` and`joined_read()` methods.
- Distinctly selected columns:
- You can specify that only distinct rows are returned by using the `%` concatenated
on to the back of a string value from the `select_params` list.
For example, if you want to return only distinct rows for the "name" column, you would use: `["name%"]`
- Distinctly selected columns can be used in conjunction with normal `select_params.` For example,
if you want to return only distinct rows for the "name" column and all rows for the "id" and "content" columns.
You would use: `["name%", "id", "content"]`
- All distinct columns must be at the beginning of the `select_params` list. If they are not, the query will fail.
#### **Example queries:**
```python
import datetime
from db.objects import db_manager
# Return the User rows with with an "ACTIVE" status and return distinct "name" values and all "id" and values.
users: dict = db_manager.read(
"User", {"status": "ACTIVE"}, ["name%", "id",]
)
```
### **Complex Examples**
```python
import datetime
from db.objects import db_manager
# In these examples there are three related tables: "User", "Post", "Comments" and "Likes".
# Return the distinct name and any id of all rows from the Users table that do not have a status of "DELETED" or "SUSPENDED".
# Return the full row of all Posts that have a user_id that matches any of the found Users and were created on or before October 1, 2021.
# Return the id and content of all Comments that have a post_id that matches any of the found Posts.
query_results: dict = db_manager.joined_read(
"User",
"Likes",
[{"status!=": ["DELETED", "SUSPENDED"]}, {"created_date<":datetime.date(year=2021, month=10, day=1)}, {}],
[["name%", "id"], [], ["id", "content"]]
)
```
Raw data
{
"_id": null,
"home_page": "",
"name": "alphonse",
"maintainer": "Jackson Prince",
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": "",
"keywords": "python,sqlalchemy",
"author": "Jackson Prince",
"author_email": "",
"download_url": "https://files.pythonhosted.org/packages/54/e1/8c90b29283c980fb69bd429b1971db1316414a72899681dfd99d4cda66b3/alphonse-0.1.0.tar.gz",
"platform": null,
"description": "# Alphonse Database Manager\n\n\n## What is?\n\nAlphonse is an interface that makes it easier to manage databases using SQLAlchemy.\nIt provides a set of methods that make complex database queries more performant and easier to write.\nFor a project already leveraging SQLAlchemy, it only requires a few lines of code to start using it immediately.\n\n\n---\n\n\n## Getting started\n\n\n### Requirements\n\n- SQLAlchemy version `1.4` or higher (`2.0+` for best performance)\n- A SQLAlchemy \"Engine\" object.\n - This is the connection to the database. It can be created using the `sqlalchemy.create_engine` method.\n- A list of all defined SQLAlchemy models that all inherit from the same instance of a `sqlalchemy.ext.declarative.declarative_base.`\n\n\n### Example setup\n\n```python\n\"\"\" Example setup for Alphonse Database Manager \"\"\"\n\nfrom alphonse import DbManager\nfrom sqlalchemy import create_engine\nfrom sqlalchemy.engine import Engine\n\nfrom my_models import MyModel1, MyModel2, MyModel3\n\n\n# Create the singelton instance of the engine\ndb_connection_url: str = \"postgresql://user:password@localhost:port/my_db_name\"\nengine: Engine = create_engine(db_connection_url)\n\n# Create a list of all defined models that all inherit from the same\n# instance of a \"declarative_base\" from sqlalchemy.ext.declarative\n\nmodel_list: list = [MyModel1, MyModel2, MyModel3]\n\n# Initialize the db_manager\ndb_manager: DbManager = DbManager(engine, model_list)\n```\n\n---\n\n\n## **Methods and usage**\n\nThe main interface to making queries is the singleton `DbManager` object created in the setup.\nIt provides a set of methods that make database queries more performant and easier to write.\nThis singleton instance acts as an abstraction layer to prevent circular imports and to provide\na single point of access to the database, ensuring that all sessions are managed correctly.\n\n---\n\n\n### **`create()`**\n\nThe `db_manager.create()` method is used to create new rows in the database. \nA boolean is returned to determine if the creation was successful.\n\n\nIt takes two arguments:\n- `table_key`: A string representing the model name. This will be the same as the class name of the model.\n- `req_payload`: A dictionary mapping of the columns and their values representing the row to be created.\n\n##### **Example:**\n```python\nfrom db.objects import db_manager\n\n# Create a new row in the \"User\" table\ncreation_map: dict = {\n\t\"username\": \"new_user\",\n\t\"password\": \"password123\",\n\t\"email\": \"test@testing.net\"\n}\n\ncreation_was_successful: bool = db_manager.create(\"User\", creation_map)\n# creation_was_successful = True or False depending on if the creation was successful.\n\n```\n\n#### `map()` method\nIf your table model has specific requirements for creating a new row, you can define a\n`map()` method directly on the model.\n\nThis method must take in a dictionary mapping of the columns, assign the key value pairs to the\nattributes of the model, and return a mapped instance of the model class or `None` if the mapping fails.\n\nThe returned class instance must be a valid instance of the model with all required fields populated\n(excepting nullable fields and auto-incrementing fields like primary key ids).\n\nThe `map()` method will be called automatically when the `db_manger.create()` method is used\nand does not need to be called directly.\nThis means that the `req_payload` dictionary passed to\nthe `create()` method only needs to include the values to be assigned to the model in the user-defined `map()` method.\n\n##### **Example model:**\n```python\n\"\"\" User table model. \"\"\"\n\nfrom typing import Optional\n\nfrom sqlalchemy import Column, Integer, String, VARCHAR\nfrom sqlalchemy.orm import relationship\n# The below imports are theorhetical and are not included in the package.\n# --------------------------------------------------------------------\n# Base is the instatiated singleton instance of `declarative_base` orignally imported from\n# sqlalchemy.ext.declarative. All models in this theorhetical project should inherit from this same instance.\nfrom db.objects.base import Base \nfrom utils.logger import logger\n\n\nclass User(Base):\n \"\"\" User orm model. \"\"\"\n\n __tablename__ = \"users\"\n\n id = Column(Integer, primary_key=True, nullable=False)\n name = Column(String(50), unique=True, nullable=False)\n status = Column(String(20), nullable=False)\n website = Column(VARCHAR(255), nullable=True)\n\n \n @classmethod\n def map(cls, req_payload: dict) -> Optional[\"User\"]:\n\t\"\"\"\n\tMap the `request_payload` dictionary to the User model.\n\t:param req_payload: A dictionary mapping of the columns in the table and their values representing the new row to be created.\n\t:return: A mapped instance of the User model or None if the mapping fails.\n\t\"\"\"\n\ttry:\n\t# Create a new instance of the User model.\n\t new_user: User = User(\n\t # Use the `.get()` method to safely access the dictionary values and return None as a default if the key is not found.\n\t\t name=req_payload.get(\"name\"),\n\t\t status=req_payload.get(\"status\"),\n\t\t partner_key=req_payload.get(\"partner_key\"),\n\t\t # In this example, the id column is an auto-incrementing primary key and the website column is nullable. \n # Therefore, they not required to be filled in when the row is created. If this method wasn't defined,\n # the `db_manager.create()` method would fail if the `req_payload` dictionary didn't include a `website` value.\n\t )\n\t # Return the mapped instance of the User model.\n\t return new_users\n\n\texcept (KeyError, AttributeError):\n\t\tlogger.log_error(\"Error mapping the request payload to the User model.\")\n\t\t# If the mapping fails, None is the expected return value to indicate that the mapping failed.\n\t\treturn None\n\n```\n\n---\n\n\n### **`read()`**\n\nThe read method is used to retrieve rows from a single table in the database.\nA dictionary of result(s) is returned, or None if the query fails.\n\nIt takes two required arguments and one optional argument:\n- `table_key`: A string representing the model name. This will be the same as the class name of the model.\n- `search_params`: A dictionary mapping of the columns in the table and their values.\n Represents the search criteria for the query.\n See the `search_params argument` section below for more information.\n- `select_params` An optional list of strings representing the columns to select from the db.\n Represents the filter parameters for the query.\n See the `select_params argument` section below for more information.\n\n##### -**`search_params` argument**\nThe values in the `search_params` dictionary are used to filter the results of the query.\nYou can supply a single value or a list of values for each key in this dictionary (column in the table.)\nFor example: `{\"status\": \"ACTIVE\"}` or `{\"status\": [\"ACTIVE\", \"SUSPENDED\"]}`\nSelect the rows where the status is \"ACTIVE\" vs. select the rows where the status is \"ACTIVE\" or \"SUSPENDED.\"\n\n##### -**`select_params` argument**\nThe `select_params` argument is an optional list of strings representing the columns to select from the db.\nThey can be used to return partial data from the table if only certain values are needed.\nThe strings must match the column names in the table exactly as they are defined on the models.\nIf a valid list is provided, only the columns in the list will be returned.\nReturns the full table row if the list is empty or not provided.\nFor example: `[\"id\", \"status\"]`\nWhatever the `search_params` return, only return the 'id' and 'status'\ncolumns (and their value) for any results from the queried table.\n\n#### **Example queries:**\n```python\nfrom typing import Optional\n\nfrom db.objects import db_manager\n\n# =============\n# BASIC EXAMPLE\n# =============\n\n# Return all rows from the \"User\" table with the a status of \"ACTIVE\".\nuser: Optional[dict] = db_manager.read(\n\t\"User\", {\"status\": \"ACTIVE\"}\n)\n\n# If one result is found:\n# user = {\n# \t\"id\": 1,\n# \t\"name\": \"test_user\",\n# \t\"status\": \"ACTIVE\",\n# \t\"website\": \"www.testwebsite.com\"\n# }\n\n# If multiple results are found, they will be within a list at a key of \"result.\"\n# user = {\n# \t\"result\": [\n# \t\t{\"id\": 1, \"name\": \"test_user\", \"status\": \"ACTIVE\", \"website\": \"www.testwebsite.com\"},\n# \t\t{\"id\": 55, \"name\": \"test_user_55\", \"status\": \"ACTIVE\", \"website\": None}\n# \t]\n# }\n\n# If the no rows are found meeting the criteria:\n# user = {}\n\n# If an exception was raised during the read operation\n# user = None\n\n\n# ====================================\n# EXAMPLE USING MULTIPLE SEARCH PARAMS\n# ====================================\n\n# Return all rows from the \"User\" table with the a status of \"ACTIVE\" or \"SUSPENDED\".\nuser: Optional[dict] = db_manager.read(\n\t\"User\", {\"status\": [\"ACTIVE\", \"SUSPENDED\"]}\n)\n\n# If multiple results are found,\n# user = {\n# \t\"result\": [\n# \t\t{\"id\": 1, \"name\": \"test_user\", \"status\": \"ACTIVE\", \"website\": \"www.testwebsite.com\"},\n# {\"id\": 55, \"name\": \"test_user_55\", \"status\": \"ACTIVE\", \"website\": None},\n# \t\t{\"id\": 55, \"name\": \"test_user_56\", \"status\": \"SUSPENDED\", \"website\": \"www.othertestwebsite.com\"}\n# \t]\n# }\n\n# ===========================\n# EXAMPLE USING SELECT PARAMS\n# ===========================\n\n# Return the id and status of all active users.\nuser: Optional[dict] = db_manager.read(\n\t\"User\",\n\t{\"status\": \"ACTIVE\"},\n\t[\"id\", \"status\",]\n)\n\n# If one result is found:\n# user = {\n# \t\"id\": 1,\n# \t\"status\": \"ACTIVE\",\n# }\n\n```\n\n\n---\n\n\n### **`update()`**\n\nThe update method is used to edit existing rows in the database. \nA boolean is returned to determine if the creation was successful.\n\n\nIt takes three arguments:\n- `table_key`: A string representing the model name. This will be the same as the class name of the model.\n- `search_params`: A dictionary mapping of parameters pertinent to specifying the query. Represents the search criteria for the query. All columns in the dictionary must be present in the table. See the `search_params argument` section below for more information.\n- `insert_params`: Mapped dictionary of key/value pairs corresponding to db columns to be updated.\nAll columns in the dictionary must be present in the table.\nOperations that leave orphaned rows will not be performed and will result in the operation failing.\n##### -**`search_params` argument**\nThe values in the `search_params` dictionary are used to filter the results of the query.\nYou can supply a single value or a list of values for each key in this dictionary (column in the table.)\nFor example: `{\"status\": \"ACTIVE\"}` or `{\"status\": [\"ACTIVE\", \"SUSPENDED\"]}`\nSelect the rows where the status is \"ACTIVE\" vs. select the rows where the status is \"ACTIVE\" or \"SUSPENDED.\"\n\n#### **Example queries:**\n```python\nfrom db.objects import db_manager\n\n# Find the row in the \"User\" table with the id of 1 and update the website column.\nparams_to_update: dict = {\"website\": \"www.newwebsite.com\"}\nupdate_was_successful: bool = db_manager.update(\n \"User\",\n {\"id\": 1},\n params_to_update\n)\n# update_was_successful = True or False depending on if the update was successful\n\n\n# Find the all rows in the \"User\" table with a status of \"ACTIVE\" or \"SUSPENDED\" and update the status column to \"DELETED\"\nupdate_was_successful: bool = db_manager.update(\n \"User\",\n {\"status\": [\"ACTIVE\", \"SUSPENDED\"]},\n {\"status\": \"DELETED\"}\n)\n# update_was_successful = True or False depending on if the update was successful\n```\n\n\n---\n\n\n### **`delete()`**\n\nThe delete method is used to remove existing rows from the database. \nA boolean is returned to determine if the creation was successful.\n\nIt takes two arguments:\n- `table_key`: A string representing the model name. This will be the same as the class name of the model.\n- `search_params`:A dictionary mapping of parameters pertinent to specifying the query. Represents the search criteria for the query. All columns in the dictionary must be present in the table. See the `search_params argument` section below for more information.\n##### -**`search_params` argument**\nThe values in the `search_params` dictionary are used to filter the results of the query.\nYou can supply a single value or a list of values for each key in this dictionary (column in the table.)\nFor example: `{\"status\": \"ACTIVE\"}` or `{\"status\": [\"ACTIVE\", \"SUSPENDED\"]}`\nSelect the rows where the status is \"ACTIVE\" vs. select the rows where the status is \"ACTIVE\" or \"SUSPENDED.\"\n\n#### **Example queries:**\n```python\nfrom db.objects import db_manager\n\n# Find the row in the \"User\" table with the id of 1 and delete it.\ndelete_was_successful: bool = db_manager.delete(\"User\", {\"id\": 1})\n# delete_was_successful = True or False depending on if the delete was successful\n\n# Find all row(s) in the \"User\" table with a status of \"ACTIVE\" or \"SUSPENDED\" and delete them.\ndelete_was_successful: bool = db_manager.delete(\"User\", {\"status\": [\"DELETE\", \"SUSPENDED\"]})\n# delete_was_successful = True or False depending on if the delete was successful\n```\n\n\n---\n\n\n### **`joined_read()`**\n\nThe joined_read method is used to retrieve rows from multiple tables in the database.\nA dictionary of results is returned, or None is returned if the query fails.\n\nIt takes three required arguments and one optional argument:\n- `starting_table`: A string representing the table where the read should start looking. This will be the same as the class name of the model.\n- `ending_table`: A string representing the table where the read should, inclusively, stop looking. This will be the same as the class name of the model.\n- `search_params`: This can be one of two datastructures:\n - A dictionary mapping of the columns in the starting table and their values representing the search criteria for the query.\n - A list of dictionary mappings each representing a table that will be traversed.\n Represents the search criteria for each table (in order traversed) for the query.\nSee `search_params argument` section below for more information.\n- `select_params` An optional list representing the columns to select from the db to be used as filter parameters for the query.\n This can be one of two datastructures:\n - A list of strings representing the columns to select from the starting table.\n - A list of lists containing strings representing the columns to select from each table in the order they are traversed.\nSee `select_params argument` section below for more information.\n\n#### -**search_params argument**\n- If only a single dict of `search_prams` is provided,\n the JOINS statement will find all rows from related tables with a foreign key pointing at the found of the starting table.\n For example, if the `starting_table` is the \"User\" table, the list of `search_params` could look like:\n ```python\n # In these examples there are three related tables: \"User\", \"Post\", \"Comments\" and \"Likes\".\n # A User can have many Posts, a Post can have many Comments and a comment can have many Likes.\n db_manager.joined_read(\n \"User\", \"Comments\", {\"id\": 1}\n )\n # Or\n db_manager.joined_read(\n \"User\", \"Comments\", [{\"id\": 1}]\n )\n # This reads as:\n # find the User with an 'id' of 1,\n # then find the all Posts that have a 'user_id' of 1,\n # then find all Comments that have a 'post_id' that matches any of the found Posts.\n ```\n You can also use a list of values to broaden the search criteria, just like in the `read()` method.\n ```python\n db_manager.joined_read(\n \"User\", \"Comments\", {\"status\": [\"ACTIVE\", \"SUSPENDED\"]}\n )\n # Or\n db_manager.joined_read(\n \"User\", \"Comments\", [{\"status\": [\"ACTIVE\", \"SUSPENDED\"]}]\n )\n # This reads as:\n # find the all Users with a status of \"ACTIVE\" or \"SUSPENDED\",\n # then find the all Posts that have a 'user_id's that match any of the found Users,\n # then find all Comments that have a 'post_id' that matches any of the found Posts.\n ```\n\n- If a list of these dictionaries is supplied, it must be the same length as the number of tables to be traversed\nin the order that they are traversed.\n An empty dict is supplied if no additional search criteria is needed for a table in the JOINS statement.\nFor example, if the starting table is \"User\"\n from the below examples and the ending table is \"Likes,\" the list of `search_param` would look like:\n\n```python\n # In these examples there are three related tables: \"User\", \"Post\", \"Comments\" and \"Likes\".\n # A User can have many Posts, a Post can have many Comments and a comment can have many Likes.\n\n db_manager.joined_read(\n \"User\", \"Likes\", [{\"id\": 1}, {\"title\": \"test_post\"}, {}]\n )\n # This reads as find the User with an 'id' of 1,\n # then find the Post with a 'user_id' of 1 and a 'title' of \"test_post,\"\n # then find all Likes that have a 'post_id' that matches the id(s) of the Post called \"test_post.\"\n\n```\n\n#### -**select_params argument**\n- If no `select_params` are provided, the full row of each table will be returned.\n- If only a single list of `select_params` is provided,\n the JOINS statement will only apply the filter to the first table in the JOINS statement.\n For example, if the `starting_table` is the \"User\" table from the below examples and\n a filter is applied, the list of select params would look like:\n - `[\"name\"],` or `[[\"name\"]]`\n This reads as, \"whatever the `search_params` find,\n only return the 'name' column for any results from the User table.\"\n\n- If a list of these lists is supplied, the filter is applied in order as the tables are traversed. For example:\n\n - `[[\"name\"],[],[\"id\", \"content\"]]`\n This reads as, \"whatever the `search_params` find, only return \n the 'name' column for any results from the User table,\n the all columns (or the full row) for any results from the Post table,\n and only return the 'id' and 'content' columns for any results from the Comments table.\"\n\n#### **Example queries:**\n```python\nfrom typing import Optional\n\nfrom db.objects import db_manager\n\n# In these examples there are three related tables: \"User\", \"Post\", \"Comments\" and \"Likes\".\n# A User can have many Posts, a Post can have many Comments and a comment can have many Likes.\n\n# =============\n# BASIC EXAMPLE\n# =============\n\n# Return the user with an 'id' of 1, all of the user's posts, & all post's comments.\nresult_object: Optional[dict] = db_manager.joined_read(\n\t\"User\",\n \"Comments\",\n {\"id\": 1}\n)\n# If some results are found:\n# result_object: dict = {\n# \t\"User\": [\n# {\"id\": 1, \"name\": \"test_user\", \"status\": \"ACTIVE\", \"website\": \"www.testwebsite.com\"}\n# ],\n# \"Posts\": [\n# {\"id\": 1, \"user_id\": 1, \"title\": \"test_post\", \"content\": \"This is a test post.\"},\n# \t {\"id\": 2, \"user_id\": 1, \"title\": \"test_post_2\", \"content\": \"This is a test post.\"}\n# ],\n# \"Comments\": [\n# {\"id\": 1, \"post_id\": 1, \"content\": \"This is a test comment.\"},\n# {\"id\": 2, \"post_id\": 1, \"content\": \"This is a test comment.\"},\n# {\"id\": 3, \"post_id\": 2, \"content\": \"This is a test comment.\"},\n# ]\n# }\n\n# If no results are found:\n# result_object: dict = {}\n\n# If an exception was raised during the read operation\n# result_object: dict = None\n\n\n# ===========================\n# EXAMPLE USING SELECT PARAMS\n# ===========================\n\n# Return the name of the user with an 'id' of 1, all of the user's posts, & all posts' comments.\nresult_object: Optional[dict] = db_manager.joined_read(\n\t\"User\",\n \"Comments\",\n {\"id\": 1},\n [\"name\"]\n)\n\n# If some results are found:\n# result_object: dict = {\n# \t\"User\": [{\"name\": \"test_user\"}],\n# \"Posts\": [\n# {\"id\": 1, \"user_id\": 1, \"title\": \"test_post\", \"content\": \"This is a test post.\"},\n# \t {\"id\": 2, \"user_id\": 1, \"title\": \"test_post_2\", \"content\": \"This is a test post.\"}\n# ],\n# \"Comments\": [\n# {\"id\": 1, \"post_id\": 1, \"content\": \"This is a test comment.\"},\n# {\"id\": 2, \"post_id\": 1, \"content\": \"This is a test comment.\"},\n# {\"id\": 3, \"post_id\": 2, \"content\": \"This is a test comment.\"},\n# ]\n# }\n\n# ====================================\n# EXAMPLE USING MULTIPLE SEARCH PARAMS\n# ====================================\n\n# Return the the user with the id of 1, the post belonging to the user with a title of \"test_post\", & all comments belonging to the post.\nresult_object: Optional[dict] = db_manager.joined_read(\n\t\"User\",\n \"Comments\",\n [{\"id\": 1}, {\"title\": \"test_post\"}, {}],\n)\n\n# If some results are found:\n# result_object: dict = {\n# \t\"User\": [\n# {\"id\": 1, \"name\": \"test_user\", \"status\": \"ACTIVE\", \"website\": \"www.testwebsite.com\"}\n# ],\n# \"Posts\": [\n# {\"id\": 1, \"user_id\": 1, \"title\": \"test_post\", \"content\": \"This is a test post.\"},\n# ],\n# \"Comments\": [\n# {\"id\": 1, \"post_id\": 1, \"content\": \"This is a test comment.\"},\n# {\"id\": 2, \"post_id\": 1, \"content\": \"This is a test comment.\"},\n# ]\n# }\n\n# ====================================\n# EXAMPLE USING MULTIPLE SELECT PARAMS\n# ====================================\n\n# Return the name of the user with an \"id\" of 1, full rows ofall of that user's posts, & the \"id\" and \"content\" of all posts'\"comments.\"\nresult_object: Optional[dict] = db_manager.joined_read(\n\t\"User\",\n \"Comments\",\n {\"id\": 1},\n [[\"name\"],[],[\"id\", \"content\"]]\n)\n\n# If some results are found:\n# result_object: dict = {\n# \t\"User\": [\n# {\"name\": \"test_user\"}\n# ],\n# \"Posts\": [\n# {\"id\": 1, \"user_id\": 1, \"title\": \"test_post\", \"content\": \"This is a test post.\"},\n# \t {\"id\": 2, \"user_id\": 1, \"title\": \"test_post_2\", \"content\": \"This is a test post.\"}\n# ],\n# \"Comments\": [\n# {\"id\": 1, \"content\": \"This is a test comment.\"},\n# {\"id\": 2, \"content\": \"This is a test comment.\"},\n# {\"id\": 3, \"content\": \"This is a test comment.\"},\n# ]\n# }\n\n# ===============================================================\n# EXAMPLE USING MULTIPLE SEARCH PARAMS AND MULTIPLE SELECT PARAMS\n# ===============================================================\n\n# Return the 'name' of the user with an `id` of 1, all posts belonging to the user with a title of \"test_post\", & and the 'id' and 'conent' of each comment belonging to the post.\nresult_object: Optional[dict] = db_manager.joined_read(\n\t\"User\",\n \"Comments\",\n [{\"id\": 1}, {\"title\": \"test_post\"}, {}],\n [[\"name\"],[],[\"id\", \"content\"]]\n)\n\n# If some results are found:\n# result_object: dict = {\n# \t\"User\": [\n# {\"name\": \"test_user\"}\n# ],\n# \"Posts\": [\n# {\"id\": 1, \"user_id\": 1, \"title\": \"test_post\", \"content\": \"This is a test post.\"},\n# ],\n# \"Comments\": [\n# {\"id\": 1, \"content\": \"This is a test comment.\"},\n# {\"id\": 2, \"content\": \"This is a test comment.\"},\n# {\"id\": 3, \"content\": \"This is a test comment.\"},\n# ]\n# }\n\n```\n\n\n---\n\n\n### **`count()`**\n\nThe count method is used to count existing rows that meet criteria in the database.\nA dictionary is returned with a count of the rows that meet the criteria, or None is returned if the count fails.\n\nIt takes two arguments:\n- `table_key`: A string representing the model name. This will be the same as the class name of the model.\n- `search_params`:A dictionary mapping of parameters pertinent to specifying the query. Represents the search criteria for the query. All columns in the dictionary must be present in the table. See the `search_params argument` section below for more information.\n##### -**`search_params` argument**\nThe values in the `search_params` dictionary are used to filter the results of the query.\nYou can supply a single value or a list of values for each key in this dictionary (column in the table.)\n\n#### **Example queries:**\n```python\nfrom db.objects import db_manager\n\n# Count the number of rows in the \"User\" table that have a status of \"DELETED\".\ncount: dict = db_manager.count(\"User\", {\"status\": \"DELETED\"})\n\n# Count the number of rows in the \"User\" table that have a status of \"DELETED\" or \"SUSPENDED\".\ncount: dict = db_manager.count(\"User\", {\"status\": [\"DELETED\", \"SUSPENDED\"]})\n\n\n# If no rows are found meeting the search criteria:\n# count = {\"count\": 0}\n\n# If some rows are found meeting the search criteria:\n# count = {\"count\": 5}\n\n# If an exception was raised during the count operation:\n# count = None \n```\n\n---\n\n\n### **Advanced Options**\n\nCertain methods have additional options that can be used to further specify the query.\n\n\n#### **search_params options**\n\n- Available to the `read()`, `update()`, `delete()`, `joined_read()`, and `count()` methods.\n- equality operators:\n\n - You can apply equality operators to the search parameters concatenating the operator to the end of the column name key. The valid operators are:\n - `==` for \"equals\" (default if no operator is provided)\n - `!=` for \"not equals\"\n - `<` for \"less than\"\n - `<=` for \"less than or equal to\"\n - `>` for \"greater than\"\n - `>=` for \"greater than or equal to\"\n \n For example, if you want to return all rows from the \"User\"\n table where the \"id\" is greater than 5, you would use: `{\"id>\": 5}`.\n\n - If a column name has no operator concatenated to the end of it, the operator will be used instead of the default \"==\" operator.\n - The data type of the value must be compatible with the operator used. For example, if you use the \"<\" operator, the value must be a number or a date. If the operator is not valid for the column type, the query will fail.\n - If an equality operator is used when multiple values are provided for a column,\n the operator will be applied to each value in the list.\n For example, if you use `{\"status!=\": [\"ACTIVE\", \"SUSPENDED\"]}`,\n the query will return all rows where the status is not \"ACTIVE\"\n or \"SUSPENDED.\"\n\n #### **Example queries:**\n\n```python\nimport datetime\nfrom db.objects import db_manager\n\n# Return all rows from the \"User\" table with the a status is not \"ACTIVE\" or \"SUSPENDED\".\nusers: dict = db_manager.read(\n \"User\", {\"status!=\": [\"ACTIVE\", \"SUSPENDED\"]}\n)\n\n# Delete all rows from the Users table with an \"id\" that is less than or equal to 4000 and a status of \"DELETED.\"\ndelete_was_succesful: bool = db_manager.delete(\n \"User\", {\"id<=\": 4000, \"status\": \"DELETED\"}\n)\n\n# Find all rows in the User table that were created on or before October 1, 2021\n# with a \"status\" of \"DELETED\" or \"SUSPENDED\" and update each rows' \"status\" to \"ACTIVE\".\nupdate_was_successful: bool = db_manager.update(\n \"User\",\n {\"created_date<=\": datetime.date(year=2021, month=10, day=1), \"status\": [\"DELETED\", \"SUSPENDED\"]},\n {\"status\": \"ACTIVE\"}\n)\n\n```\n\n#### **select_params options**\n\n- Available to the `read()` and`joined_read()` methods.\n- Distinctly selected columns:\n - You can specify that only distinct rows are returned by using the `%` concatenated\n on to the back of a string value from the `select_params` list.\n For example, if you want to return only distinct rows for the \"name\" column, you would use: `[\"name%\"]`\n - Distinctly selected columns can be used in conjunction with normal `select_params.` For example,\n if you want to return only distinct rows for the \"name\" column and all rows for the \"id\" and \"content\" columns.\n You would use: `[\"name%\", \"id\", \"content\"]`\n - All distinct columns must be at the beginning of the `select_params` list. If they are not, the query will fail.\n\n\n #### **Example queries:**\n\n```python\nimport datetime\nfrom db.objects import db_manager\n\n# Return the User rows with with an \"ACTIVE\" status and return distinct \"name\" values and all \"id\" and values.\nusers: dict = db_manager.read(\n \"User\", {\"status\": \"ACTIVE\"}, [\"name%\", \"id\",]\n)\n\n```\n\n\n### **Complex Examples**\n```python\nimport datetime\nfrom db.objects import db_manager\n\n# In these examples there are three related tables: \"User\", \"Post\", \"Comments\" and \"Likes\".\n\n# Return the distinct name and any id of all rows from the Users table that do not have a status of \"DELETED\" or \"SUSPENDED\".\n# Return the full row of all Posts that have a user_id that matches any of the found Users and were created on or before October 1, 2021.\n# Return the id and content of all Comments that have a post_id that matches any of the found Posts.\nquery_results: dict = db_manager.joined_read(\n \"User\",\n \"Likes\",\n [{\"status!=\": [\"DELETED\", \"SUSPENDED\"]}, {\"created_date<\":datetime.date(year=2021, month=10, day=1)}, {}],\n [[\"name%\", \"id\"], [], [\"id\", \"content\"]]\n)\n\n```\n",
"bugtrack_url": null,
"license": "Copyright (c) 2024 Wavelynx Technologies Corporation Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the \"Software\"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED \"AS IS\", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. ",
"summary": "A wrapper around SQLAlchemy to make it easier to make more complex queries",
"version": "0.1.0",
"project_urls": {
"Source": "https://gitlab.com/WaveLynxTech/alphonse"
},
"split_keywords": [
"python",
"sqlalchemy"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "f05027c029c553962e9b0427cc0381a14cb80335fbd433604d3b7d8ca320f866",
"md5": "6843220729e50dcf7ab4707be73af6c9",
"sha256": "b1820d122f4016d9a6d412676f074947cc12a81f294e369c2451c6433b7aeca6"
},
"downloads": -1,
"filename": "alphonse-0.1.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "6843220729e50dcf7ab4707be73af6c9",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 24840,
"upload_time": "2024-02-13T19:48:43",
"upload_time_iso_8601": "2024-02-13T19:48:43.517294Z",
"url": "https://files.pythonhosted.org/packages/f0/50/27c029c553962e9b0427cc0381a14cb80335fbd433604d3b7d8ca320f866/alphonse-0.1.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "54e18c90b29283c980fb69bd429b1971db1316414a72899681dfd99d4cda66b3",
"md5": "fc6e7b37c0ad15a919c7116dba894938",
"sha256": "b961d295646bfb3254f9bc5e65e094fd260340b8510bf662621da0c2fc8fcbd5"
},
"downloads": -1,
"filename": "alphonse-0.1.0.tar.gz",
"has_sig": false,
"md5_digest": "fc6e7b37c0ad15a919c7116dba894938",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 35536,
"upload_time": "2024-02-13T19:48:45",
"upload_time_iso_8601": "2024-02-13T19:48:45.839035Z",
"url": "https://files.pythonhosted.org/packages/54/e1/8c90b29283c980fb69bd429b1971db1316414a72899681dfd99d4cda66b3/alphonse-0.1.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-02-13 19:48:45",
"github": false,
"gitlab": true,
"bitbucket": false,
"codeberg": false,
"gitlab_user": "WaveLynxTech",
"gitlab_project": "alphonse",
"lcname": "alphonse"
}
