Security News
tea.xyz Spam Plagues npm and RubyGems Package Registries
Tea.xyz, a crypto project aimed at rewarding open source contributions, is once again facing backlash due to an influx of spam packages flooding public package registries.
easyquery-query-builder
Readme
Main purpose of package is to provide user with ability to create sql queries using QueryBuilders which are implemented in Builder Project Pattern.
Using pip:
pip install easyquery-query-builder
Using poetry:
poetry add easyquery-query-builder
Using pipenv:
pipenv install easyquery-query-builder
All functionalities are fully tested.
You are able to run tests on your own by using cloning repository and using it's virtual environment.
git clone https://github.com/DSmolke/EASYQUERY_QUERY_BUILDER.git
cd EASYQUERY_QUERY_BUILDER
pip install pytest
pip install coverage
pip install easyvalid-data-validator
pytest -vv
pip install poetry
poetry update
poetry shell
poetry run pytest -vv
(poetry update is used due to some issue that I faced with newest Python version)
pip install pipenv
pipenv shell
pipenv run pytest -vv
If you haven't done step 1, start from here:
git clone https://github.com/DSmolke/EASYQUERY_QUERY_BUILDER.git
cd EASYQUERY_QUERY_BUILDER
Then use this command to change branch
git checkout with_coverage_files
You should be able to see htmlcov directory. Enter it and open index.html file to see full coverage report.
Let's say we need to prepare query where we ask database for all records from drivers table, joined with licenses table on license id:
from easyquery_query_builder.queries.read_query_with_joins import ReadQueryWithJoins
from easyquery_query_builder.queries.read_query_with_joins_builder import ReadQueryWithJoinsBuilder
query: ReadQueryWithJoins = ReadQueryWithJoinBuilder() \
.add_select_statement('*') \
.add_from_statement('drivers') \
.add_joins_statement([['licenses', 'l', 'l.id = drivers.license_id']])\
.build()
Once we built our query it's ready to be used:
print(query.parse())
with result of:
select * from drivers join licenses as l on l.id = drivers.license_id
At this stage of development one abstract class and two inheriting classes are implemented:
Contains one abstract method .parse() which force to take no arguments and return string value. It will be used for final step in query creation.
class Query(ABC):
@abstractmethod
def parse(self) -> str:
pass
It's purpose is to create and manage sql query that will be returned using parse method.
class ReadQuery:
def __init__(self, select_="", from_="", where_="", group_by_="", having_="", order_by_=""):
""" Empty query is created if no values are provided.(designed for builder)"""
self.select_ = select_
self.from_ = from_
self.where_ = where_
self.group_by_ = group_by_
self.having_ = having_
self.order_by_ = order_by_
As you can see, all statements like select, from etc. have sufix of '_'. It helps with modern IDE like PyCharm to hide sql syntax which can be targeted and indicate fake errors occurring in the code.
def parse(self) -> str:
""" Creates sql query expression using fields provided in instance """
# validation of all fields - checks only types
query_data = self.__dict__
constraints = {
"select_": {Constraint.IS_TYPE: str},
"from_": {Constraint.IS_TYPE: str},
"where_": {Constraint.IS_TYPE: str},
"group_by_": {Constraint.IS_TYPE: str},
"having_": {Constraint.IS_TYPE: str},
"order_by_": {Constraint.IS_TYPE: str}
}
validate_json_data(query_data, constraints)
# validation that force all fields to be properly structured - minimum of select and from statements, having only with group by
s, f, w, g, h = self.select_, self.from_, self.where_, self.group_by_, self.having_
if (s == "" and f == "") or (s != "" and f == "") or (s == "" and f != ""):
raise ValueError("Query requirement is to have select and from statements")
if h != "" and g == "":
raise ValueError("You cannot use having block without declaring group by block")
# creation of sql query
statement = f"{f'select {self.select_}' if self.select_ else ''}" \
f"{f' from {self.from_}' if self.from_ else ''}" \
f"{f' where {self.where_}' if self.where_ else ''}" \
f"{f' group by {self.group_by_}' if self.group_by_ else ''}" \
f"{f' having {self.having_}' if self.having_ else ''}" \
f"{f' order by {self.order_by_}' if self.order_by_ else ''}"
return statement
Class that inheriting ReadQuery, there are two minor differences between them:
class ReadQueryWithJoins(ReadQuery):
""" Subclass of ReadQuery which implements joins """
def __init__(self, select_="", from_="", where_="", group_by_="", having_="", order_by_="", joins_=""):
super().__init__(select_, from_, where_, group_by_, having_, order_by_)
self.joins_: list[list[str] | str] = joins_
filed joins_ added
def parse(self) -> str:
""" Creates sql query expression can be extended with join statements using fields provided in instance """
# validation of all fields - checks only types and types of members
query_data = self.__dict__
constraints = {
"select_": {Constraint.IS_TYPE: str},
"from_": {Constraint.IS_TYPE: str},
"where_": {Constraint.IS_TYPE: str},
"group_by_": {Constraint.IS_TYPE: str},
"having_": {Constraint.IS_TYPE: str},
"order_by_": {Constraint.IS_TYPE: str},
"joins_": {Constraint.IS_TYPE: list, Constraint.ARRAY_MEMBERS_TYPE: list}
}
validate_json_data(query_data, constraints)
# validation that force all fields to be properly structured - minimum of select and from statements, having only with group by
s, f, w, g, h = self.select_, self.from_, self.where_, self.group_by_, self.having_
if (s == "" and f == "") or (s != "" and f == "") or (s == "" and f != ""):
raise ValueError("Query requirement is to have select and from statements")
if h != "" and g == "":
raise ValueError("You cannot use having block without declaring group by block")
# concatenation of joins
joins_exp = " ".join([f"join {table} as {alias} on {conditions}" for table, alias, conditions in self.joins_])
# creation of sql query
statement = f"{f'select {self.select_}' if self.select_ else ''}" \
f"{f' from {self.from_}' if self.from_ else ''}" \
f" {joins_exp}" \
f"{f' where {self.where_}' if self.where_ else ''}" \
f"{f' group by {self.group_by_}' if self.group_by_ else ''}" \
f"{f' having {self.having_}' if self.having_ else ''}" \
f"{f' order by' if self.order_by_ else ''}"
return statement
"joins_": {Constraint.IS_TYPE: list, Constraint.ARRAY_MEMBERS_TYPE: list}
Added to constraints and validated, as well as jons_exp now is a part of sql statement.
Their role is to create new or manage existing queries
class ReadQueryBuilder:
""" Builder used to create new ReadQueries 'from scratch' or modify existing ones to desired form """
def __init__(self, query: Query | None = None):
if query is None:
self.query = ReadQuery()
else:
self.query = query
At this stage, new ReadQuery is made if no query is provided as an argument.
User can change state of query at any point by using one of method that has same structure as this one:
def add_select_statement(self, new_select: str) -> Self:
""" Ads new select statement provided by user. Basic validation of argument is performed"""
data = {"new_select": new_select}
constraints = {"new_select": {Constraint.IS_TYPE: str}}
validate_json_data(data, constraints)
self.query.select_ = new_select
return self
It allows us to chain methods:
ReadQueryBuilder().add_select_statement('*').add_from_statement('cars').add_where_statement('production_year > 2020')
and
Basically returns query that was created or managed by builder:
def build(self) -> ReadQuery:
""" Builds query based on all operations that were made """
return self.query
Inherits ReadQueryBuilder but operates on ReadQueryWithJoins:
def __init__(self, query: Query | None = None):
if query is None:
self.query = ReadQueryWithJoins()
else:
self.query = query
It has all the futures of ReadQueryBuilder but adds method for joins:
def add_joins_statement(self, new_joins: str) -> Self:
""" Ads new joins arguments provided by user. Basic validation of argument is performed"""
data = {"new_joins": new_joins}
constraints = {"new_joins": {Constraint.IS_TYPE: list, Constraint.ARRAY_MEMBERS_TYPE: list}}
validate_json_data(data, constraints)
self.query.joins_ = new_joins
return self
and
Basically returns query that was created or managed by builder:
def build(self) -> ReadQueryWithJoins:
""" Builds query based on all operations that were made """
return self.query
ReadQueryBuilder().add_select_statement('id, age')
ReadQueryBuilder().add_select_statement('*')
ReadQueryBuilder().add_select_statement('cars.id, owner.*')
query = ReadQueryBuilder().add_select_statement('*').build()
query.parse()
File "<some path....>", line 47, in parse
raise ValueError("Query requirement is to have select and from statements")
ValueError: Query requirement is to have select and from statements
query = ReadQueryBuilder()\
.add_select_statement('*')\
.add_from_statement('cars')\
.add_having_statement("value > 300000")\
.build()
query.parse()
File "<some path....>", line 49, in parse
raise ValueError("You cannot use having block without declaring group by block")
ValueError: You cannot use having block without declaring group by block
query = ReadQueryWithJoinsBuilder()\
.add_select_statement('*')\
.add_from_statement('cars')\
.add_joins_statement([['drivers', 'd', 'd.id = cars.driver_id'], ['producers', 'p', 'p.id = cars.prod_id']])\
.build()
query.parse()
Outcome:
select * from cars join drivers as d on d.id = cars.driver_id join producers as p on p.id = cars.prod_id
query = ReadQueryWithJoinsBuilder()\
.add_select_statement(100)\
.add_from_statement('cars')
query = ReadQueryWithJoinsBuilder().add_select_statement(100)
File "<some path....>", line 121, in validate_json_data
raise ValidationError(dict(errors_by_key))
easyvalid_data_validator.customexceptions.common.ValidationError: {'new_select': ["Invalid type - isn't same type like compare type"]}
or in parse:
sql_expression = ReadQuery(select_=100).parse()
File File "<some path....>", line 121, in validate_json_data
raise ValidationError(dict(errors_by_key))
easyvalid_data_validator.customexceptions.common.ValidationError: {'select_': ["Invalid type - isn't same type like compare type"]}
FAQs
Unknown package
We found that easyquery-query-builder demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
Tea.xyz, a crypto project aimed at rewarding open source contributions, is once again facing backlash due to an influx of spam packages flooding public package registries.
Security News
As cyber threats become more autonomous, AI-powered defenses are crucial for businesses to stay ahead of attackers who can exploit software vulnerabilities at scale.
Security News
UnitedHealth Group disclosed that the ransomware attack on Change Healthcare compromised protected health information for millions in the U.S., with estimated costs to the company expected to reach $1 billion.