Compare commits

..

5 Commits

Author SHA1 Message Date
renovate[bot]
79dde8008a feat(deps): update dependency stripe to v12 (#42)
Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com>
2025-04-09 14:01:08 +02:00
creyD
adb017d6ce Adjusted files for isort & autopep 2025-04-09 12:01:00 +00:00
vikynoah
e160cc5fea feat: Response filter async (#47)
* fix: get_object alter for async response filter

* fix: Alter async response

* feat: Decorator for schema out
2025-04-09 14:00:30 +02:00
creyD
7afb8e2fd8 Adjusted files for isort & autopep 2025-04-04 15:55:18 +00:00
vikynoah
badf2b157f Response filter async (#45)
* fix: get_object alter for async response filter

* fix: Alter async response
2025-04-04 17:54:47 +02:00
5 changed files with 31 additions and 9 deletions

View File

@@ -1,12 +1,13 @@
from typing import Type, TypeVar, overload, List
import asyncio
from typing import List, Type, TypeVar, overload
from uuid import UUID
from fastapi import HTTPException
from pydantic import BaseModel
from sqlalchemy.orm import Session
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
import asyncio
from sqlalchemy.orm import Session
from .models.base import Base
T = TypeVar("T", bound=Base)
@@ -64,11 +65,9 @@ def get_object_or_404(
query = select(db_class).where(getattr(db_class, lookup_column) == id)
result = await db.execute(query)
row = result.scalar_one_or_none()
if row is None:
raise HTTPException(status_code=404, detail="The object does not exist.")
obj_dict = {k: v for k, v in row.__dict__.items() if not k.startswith("_")}
obj_dict = row
if expunge:
await db.expunge(obj_dict)
return obj_dict

View File

@@ -1,2 +1,3 @@
from .base import * # noqa
from .response_schema import * #noqa
from .schema_optional import * #noqa

View File

@@ -1,6 +1,7 @@
from typing import List, Optional, Type
from pydantic import BaseModel, create_model
from fastapi import Query
from pydantic import BaseModel, create_model
class ResponseModelDependency:
@@ -8,8 +9,10 @@ class ResponseModelDependency:
self.model_class = model_class
def __call__(self, response_fields: Optional[List[str]] = Query(None)) -> Type[BaseModel]:
def process_result(result, fields=None):
def process_result(result, fields=None, async_session=False):
if not fields:
if async_session:
return {k: v for k, v in result.__dict__.items() if not k.startswith("_")}
return result
if hasattr(result, "_fields"):

View File

@@ -0,0 +1,19 @@
from typing import Optional, Type, Union, get_args, get_origin, get_type_hints
from pydantic import BaseModel, create_model
def optional_fields(cls: Type[BaseModel]) -> Type[BaseModel]:
fields = {}
for name, hint in get_type_hints(cls).items():
if name.startswith("_"):
continue
if get_origin(hint) is not Union or type(None) not in get_args(hint):
hint = Optional[hint]
fields[name] = (hint, None)
new_model = create_model(cls.__name__, __base__=cls, **fields)
return new_model

View File

@@ -1 +1 @@
stripe==11.6.0 # Stripe
stripe==12.0.0 # Stripe