Compare commits

...

8 Commits
0.2.1 ... 1.2.4

6 changed files with 121 additions and 45 deletions

View File

@@ -2,6 +2,22 @@
All notable changes to this project will be documented in this file.
## 1.2.4
- Enabled newer versions for all dependencies
## 1.2.3
- Added option to specify lookup_column for get_object_or_404
## 1.2.2
- Added order_by method
## 1.1.0
- Added headers to testing
## 1.0.0
- Bumped dependencies

View File

@@ -55,3 +55,8 @@ from creyPY.const import LanguageEnum
print(LanguageEnum.EN) # Output: LanguageEnum.EN
print(LanguageEnum.EN.value) # Output: English
```
## TODO
- Add async support for database connection
- Add version without postgresql dependency

View File

@@ -10,8 +10,10 @@ from .models.base import Base
T = TypeVar("T", bound=Base)
def get_object_or_404(db_class: Type[T], id: UUID | str, db: Session, expunge: bool = False) -> T:
obj = db.query(db_class).filter(db_class.id == id).one_or_none()
def get_object_or_404(
db_class: Type[T], id: UUID | str, db: Session, expunge: bool = False, lookup_column: str = "id"
) -> T:
obj = db.query(db_class).filter(getattr(db_class, lookup_column) == id).one_or_none()
if obj is None:
raise HTTPException(status_code=404, detail="The object does not exist.")
if expunge:
@@ -36,7 +38,7 @@ def update_obj_from_data(
model: Type[T],
id: UUID | str,
db: Session,
partial: bool = False,
partial: bool = False, # TODO: inverse, because it is currently the wrong way around
ignore_fields=[],
additional_data={},
exclude={},

View File

@@ -0,0 +1,25 @@
from typing import Callable
from pydantic.json_schema import SkipJsonSchema
from sqlalchemy import String, asc, cast, desc
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.sql.selectable import Select
def order_by(order_by: str | SkipJsonSchema[None] = None) -> Callable[[Select], Select]:
def _order_by(query: Select) -> Select:
if order_by:
direction = desc if order_by.startswith("-") else asc
column_name = order_by.lstrip("-")
# Get the column from the query
for column in query.inner_columns:
if column.key == column_name:
# If the column is a UUID, cast it to a string
if isinstance(column.type, UUID):
column = cast(column, String)
query = query.order_by(direction(column))
break
return query
return _order_by

View File

@@ -6,26 +6,31 @@ from fastapi.testclient import TestClient
class GenericClient(TestClient):
def __init__(self, client):
self.c = TestClient(client)
self.default_headers = {}
def get(self, url: str, r_code: int = 200, parse_json=True):
re = self.c.get(url)
re = self.c.get(url, headers=self.default_headers)
if re.status_code != r_code:
print(re.content)
assert r_code == re.status_code
return re.json() if parse_json else re.content
def delete(self, url: str, r_code: int = 204):
re = self.c.delete(url)
re = self.c.delete(url, headers=self.default_headers)
if re.status_code != r_code:
print(re.content)
assert r_code == re.status_code
return re.json() if r_code != 204 else None
def post(
self, url: str, obj: dict = {}, r_code: int = 201, raw_response=False, *args, **kwargs
self, url: str, obj: dict | str = {}, r_code: int = 201, raw_response=False, *args, **kwargs
):
re = self.c.post(
url, data=json.dumps(obj), headers={"Content-Type": "application/json"}, *args, **kwargs
url,
data=json.dumps(obj) if type(obj) == dict else obj,
headers=self.default_headers | {"Content-Type": "application/json"},
*args,
**kwargs,
)
if re.status_code != r_code:
print(re.content)
@@ -33,28 +38,41 @@ class GenericClient(TestClient):
return re.json() if not raw_response else re
def post_file(self, url: str, file, r_code: int = 201, raw_response=False, *args, **kwargs):
re = self.c.post(url, files={"file": file}, *args, **kwargs)
if re.status_code != r_code:
print(re.content)
assert r_code == re.status_code
return re.json() if not raw_response else re
def patch(
self, url: str, obj: dict = {}, r_code: int = 200, raw_response=False, *args, **kwargs
):
re = self.c.patch(
url, data=json.dumps(obj), headers={"Content-Type": "application/json"}, *args, **kwargs
re = self.c.post(
url,
files={"file": file},
headers=self.default_headers | {"Content-Type": "application/json"},
*args,
**kwargs,
)
if re.status_code != r_code:
print(re.content)
assert r_code == re.status_code
return re.json() if not raw_response else re
def put(self, url: str, obj: dict = {}, r_code: int = 200, raw_response=False, *args, **kwargs):
def patch(
self, url: str, obj: dict | str = {}, r_code: int = 200, raw_response=False, *args, **kwargs
):
re = self.c.patch(
url,
data=json.dumps(obj) if type(obj) == dict else obj,
headers=self.default_headers | {"Content-Type": "application/json"},
*args,
**kwargs,
)
if re.status_code != r_code:
print(re.content)
assert r_code == re.status_code
return re.json() if not raw_response else re
def put(
self, url: str, obj: dict | str = {}, r_code: int = 200, raw_response=False, *args, **kwargs
):
re = self.c.put(
url,
data=json.dumps(obj),
headers={
data=json.dumps(obj) if type(obj) == dict else obj,
headers=self.default_headers
| {
"Content-Type": "application/json",
"accept": "application/json",
},
@@ -66,7 +84,14 @@ class GenericClient(TestClient):
assert r_code == re.status_code
return re.json() if not raw_response else re
def obj_lifecycle(self, input_obj: dict, url: str, pagination: bool = True):
def obj_lifecycle(
self,
input_obj: dict,
url: str,
pagination: bool = True,
id_field: str = "id",
created_at_check: bool = True,
):
# GET LIST
re = self.get(url)
if pagination:
@@ -77,15 +102,18 @@ class GenericClient(TestClient):
# CREATE
re = self.post(url, obj=input_obj)
assert "id" in re
assert re["id"] is not None
assert re["created_at"] is not None
assert id_field in re
assert re[id_field] is not None
obj_id = str(re["id"])
if created_at_check:
assert "created_at" in re
assert re["created_at"] is not None
obj_id = str(re[id_field])
# GET
re = self.get(f"{url}{obj_id}/")
assert re["id"] == obj_id
assert re[id_field] == obj_id
# GET LIST
re = self.get(url)

View File

@@ -1,23 +1,23 @@
annotated-types==0.6.0 # Pydantic
pydantic==2.6.4 # Pydantic
pydantic-core==2.16.3 # Pydantic
typing-extensions==4.10.0 # Pydantic
annotated-types>=0.6.0 # Pydantic
pydantic>=2.6.4 # Pydantic
pydantic-core>=2.16.3 # Pydantic
typing-extensions>=4.10.0 # Pydantic
anyio==4.3.0 # FastAPI
fastapi==0.110.1 # FastAPI
idna==3.6 # FastAPI
sniffio==1.3.1 # FastAPI
starlette==0.37.2 # FastAPI
anyio>=4.3.0 # FastAPI
fastapi>=0.110.1 # FastAPI
idna>=3.6 # FastAPI
sniffio>=1.3.1 # FastAPI
starlette>=0.37.2 # FastAPI
fastapi-pagination==0.12.21 # Pagination
sqlalchemy==2.0.29 # SQLAlchemy
fastapi-pagination>=0.12.21 # Pagination
sqlalchemy>=2.0.29 # SQLAlchemy
python-dotenv==1.0.1 # Environment variables
python-dotenv>=1.0.1 # Environment variables
psycopg==3.1.18 # PostgreSQL
psycopg-binary==3.1.18 # PostgreSQL
psycopg-pool==3.2.1 # PostgreSQL
psycopg>=3.1.18 # PostgreSQL
psycopg-binary>=3.1.18 # PostgreSQL
psycopg-pool>=3.2.1 # PostgreSQL
h11==0.14.0 # Testing
httpcore==1.0.5 # Testing
httpx==0.27.0 # Testing
h11>=0.14.0 # Testing
httpcore>=1.0.5 # Testing
httpx>=0.27.0 # Testing