mirror of
https://github.com/creyD/creyPY.git
synced 2026-04-13 11:50:31 +02:00
Compare commits
7 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 3f0379290d | |||
|
|
fa7a1c8a61 | ||
| 5a7e1776db | |||
| 4c25d91e18 | |||
| f24db62781 | |||
| 4d997a375e | |||
| ee11d86235 |
16
CHANGELOG.md
16
CHANGELOG.md
@@ -2,6 +2,22 @@
|
|||||||
|
|
||||||
All notable changes to this project will be documented in this file.
|
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
|
## 1.0.0
|
||||||
|
|
||||||
- Bumped dependencies
|
- Bumped dependencies
|
||||||
|
|||||||
@@ -56,7 +56,7 @@ print(LanguageEnum.EN) # Output: LanguageEnum.EN
|
|||||||
print(LanguageEnum.EN.value) # Output: English
|
print(LanguageEnum.EN.value) # Output: English
|
||||||
```
|
```
|
||||||
|
|
||||||
|
|
||||||
## TODO
|
## TODO
|
||||||
|
|
||||||
- Add async support for database connection
|
- Add async support for database connection
|
||||||
|
- Add version without postgresql dependency
|
||||||
|
|||||||
@@ -10,8 +10,10 @@ from .models.base import Base
|
|||||||
T = TypeVar("T", bound=Base)
|
T = TypeVar("T", bound=Base)
|
||||||
|
|
||||||
|
|
||||||
def get_object_or_404(db_class: Type[T], id: UUID | str, db: Session, expunge: bool = False) -> T:
|
def get_object_or_404(
|
||||||
obj = db.query(db_class).filter(db_class.id == id).one_or_none()
|
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:
|
if obj is None:
|
||||||
raise HTTPException(status_code=404, detail="The object does not exist.")
|
raise HTTPException(status_code=404, detail="The object does not exist.")
|
||||||
if expunge:
|
if expunge:
|
||||||
@@ -36,7 +38,7 @@ def update_obj_from_data(
|
|||||||
model: Type[T],
|
model: Type[T],
|
||||||
id: UUID | str,
|
id: UUID | str,
|
||||||
db: Session,
|
db: Session,
|
||||||
partial: bool = False,
|
partial: bool = False, # TODO: inverse, because it is currently the wrong way around
|
||||||
ignore_fields=[],
|
ignore_fields=[],
|
||||||
additional_data={},
|
additional_data={},
|
||||||
exclude={},
|
exclude={},
|
||||||
|
|||||||
25
creyPY/fastapi/order_by.py
Normal file
25
creyPY/fastapi/order_by.py
Normal 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
|
||||||
@@ -6,26 +6,31 @@ from fastapi.testclient import TestClient
|
|||||||
class GenericClient(TestClient):
|
class GenericClient(TestClient):
|
||||||
def __init__(self, client):
|
def __init__(self, client):
|
||||||
self.c = TestClient(client)
|
self.c = TestClient(client)
|
||||||
|
self.default_headers = {}
|
||||||
|
|
||||||
def get(self, url: str, r_code: int = 200, parse_json=True):
|
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:
|
if re.status_code != r_code:
|
||||||
print(re.content)
|
print(re.content)
|
||||||
assert r_code == re.status_code
|
assert r_code == re.status_code
|
||||||
return re.json() if parse_json else re.content
|
return re.json() if parse_json else re.content
|
||||||
|
|
||||||
def delete(self, url: str, r_code: int = 204):
|
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:
|
if re.status_code != r_code:
|
||||||
print(re.content)
|
print(re.content)
|
||||||
assert r_code == re.status_code
|
assert r_code == re.status_code
|
||||||
return re.json() if r_code != 204 else None
|
return re.json() if r_code != 204 else None
|
||||||
|
|
||||||
def post(
|
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(
|
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:
|
if re.status_code != r_code:
|
||||||
print(re.content)
|
print(re.content)
|
||||||
@@ -33,28 +38,41 @@ class GenericClient(TestClient):
|
|||||||
return re.json() if not raw_response else re
|
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):
|
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)
|
re = self.c.post(
|
||||||
if re.status_code != r_code:
|
url,
|
||||||
print(re.content)
|
files={"file": file},
|
||||||
assert r_code == re.status_code
|
headers=self.default_headers | {"Content-Type": "application/json"},
|
||||||
return re.json() if not raw_response else re
|
*args,
|
||||||
|
**kwargs,
|
||||||
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
|
|
||||||
)
|
)
|
||||||
if re.status_code != r_code:
|
if re.status_code != r_code:
|
||||||
print(re.content)
|
print(re.content)
|
||||||
assert r_code == re.status_code
|
assert r_code == re.status_code
|
||||||
return re.json() if not raw_response else re
|
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(
|
re = self.c.put(
|
||||||
url,
|
url,
|
||||||
data=json.dumps(obj),
|
data=json.dumps(obj) if type(obj) == dict else obj,
|
||||||
headers={
|
headers=self.default_headers
|
||||||
|
| {
|
||||||
"Content-Type": "application/json",
|
"Content-Type": "application/json",
|
||||||
"accept": "application/json",
|
"accept": "application/json",
|
||||||
},
|
},
|
||||||
@@ -66,7 +84,14 @@ class GenericClient(TestClient):
|
|||||||
assert r_code == re.status_code
|
assert r_code == re.status_code
|
||||||
return re.json() if not raw_response else re
|
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
|
# GET LIST
|
||||||
re = self.get(url)
|
re = self.get(url)
|
||||||
if pagination:
|
if pagination:
|
||||||
@@ -77,15 +102,18 @@ class GenericClient(TestClient):
|
|||||||
|
|
||||||
# CREATE
|
# CREATE
|
||||||
re = self.post(url, obj=input_obj)
|
re = self.post(url, obj=input_obj)
|
||||||
assert "id" in re
|
assert id_field in re
|
||||||
assert re["id"] is not None
|
assert re[id_field] is not None
|
||||||
assert re["created_at"] 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
|
# GET
|
||||||
re = self.get(f"{url}{obj_id}/")
|
re = self.get(f"{url}{obj_id}/")
|
||||||
assert re["id"] == obj_id
|
assert re[id_field] == obj_id
|
||||||
|
|
||||||
# GET LIST
|
# GET LIST
|
||||||
re = self.get(url)
|
re = self.get(url)
|
||||||
|
|||||||
@@ -1,23 +1,23 @@
|
|||||||
annotated-types==0.6.0 # Pydantic
|
annotated-types>=0.6.0 # Pydantic
|
||||||
pydantic==2.6.4 # Pydantic
|
pydantic>=2.6.4 # Pydantic
|
||||||
pydantic-core==2.16.3 # Pydantic
|
pydantic-core>=2.16.3 # Pydantic
|
||||||
typing-extensions==4.10.0 # Pydantic
|
typing-extensions>=4.10.0 # Pydantic
|
||||||
|
|
||||||
anyio==4.3.0 # FastAPI
|
anyio>=4.3.0 # FastAPI
|
||||||
fastapi==0.110.1 # FastAPI
|
fastapi>=0.110.1 # FastAPI
|
||||||
idna==3.6 # FastAPI
|
idna>=3.6 # FastAPI
|
||||||
sniffio==1.3.1 # FastAPI
|
sniffio>=1.3.1 # FastAPI
|
||||||
starlette==0.37.2 # FastAPI
|
starlette>=0.37.2 # FastAPI
|
||||||
|
|
||||||
fastapi-pagination==0.12.21 # Pagination
|
fastapi-pagination>=0.12.21 # Pagination
|
||||||
sqlalchemy==2.0.29 # SQLAlchemy
|
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>=3.1.18 # PostgreSQL
|
||||||
psycopg-binary==3.1.18 # PostgreSQL
|
psycopg-binary>=3.1.18 # PostgreSQL
|
||||||
psycopg-pool==3.2.1 # PostgreSQL
|
psycopg-pool>=3.2.1 # PostgreSQL
|
||||||
|
|
||||||
h11==0.14.0 # Testing
|
h11>=0.14.0 # Testing
|
||||||
httpcore==1.0.5 # Testing
|
httpcore>=1.0.5 # Testing
|
||||||
httpx==0.27.0 # Testing
|
httpx>=0.27.0 # Testing
|
||||||
|
|||||||
Reference in New Issue
Block a user