diff --git a/pyproject.toml b/pyproject.toml index 6663a6c..daaf034 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,6 +14,7 @@ dependencies = [ "django-ninja>=0.22", "pydantic>=2.0", "inflect>=7.5", + "colorama>=0.4.6" ] classifiers = [ diff --git a/src/lazy_ninja/helpers.py b/src/lazy_ninja/helpers.py index f7e5973..f00ec24 100644 --- a/src/lazy_ninja/helpers.py +++ b/src/lazy_ninja/helpers.py @@ -4,9 +4,21 @@ from django.db.models import QuerySet, Model from django.core.exceptions import FieldDoesNotExist +from django.db import models from ninja import Schema +def parse_model_id(model: Type[models.Model], item_id: str) -> Any: + """ + Converts a path parameter string to the correct type for the model's PK. + Supports AutoField (int) and UUIDField (str). + """ + pk_field = model._meta.pk + if isinstance(pk_field, models.AutoField) and item_id.isdigit(): + return int(item_id) + return item_id + + def get_hook(controller: Optional[Any], hook_name: str) -> Optional[Callable]: """ Safely get a hook method from a controller. diff --git a/src/lazy_ninja/routes.py b/src/lazy_ninja/routes.py index 6758614..ba9f6d2 100644 --- a/src/lazy_ninja/routes.py +++ b/src/lazy_ninja/routes.py @@ -18,7 +18,7 @@ execute_hook_async, ) from .pagination import BasePagination -from .helpers import execute_hook, handle_response, apply_filters, apply_filters_async +from .helpers import execute_hook, handle_response, apply_filters, apply_filters_async, parse_model_id from .errors import handle_exception, handle_exception_async from .file_upload import FileUploadConfig, detect_file_fields @@ -107,10 +107,11 @@ async def list_items( return await handle_exception_async(e) @router.get("/{item_id}", response=detail_schema, tags=[model.__name__], operation_id=f"get_{model_name}") - async def get_item(request, item_id: int) -> Any: + async def get_item(request, item_id: str) -> Any: """Retrieve a single object by ID.""" try: - instance = await get_object_or_404_async(model, id=item_id) + item_id_value = parse_model_id(model, item_id) + instance = await get_object_or_404_async(model, id=item_id_value) return await handle_response_async(instance, detail_schema, custom_response, request) except Exception as e: return await handle_exception_async(e) @@ -238,10 +239,11 @@ async def create_item(request, payload: create_schema) -> Any: #type: ignore if update_schema: if use_multipart_update: @router.patch("/{item_id}", response=detail_schema, tags=[model.__name__], operation_id=f"update_{model_name}") - async def update_item(request, item_id: int, payload: update_schema = Form(...)) -> Any: #type: ignore + async def update_item(request, item_id: str, payload: update_schema = Form(...)) -> Any: #type: ignore """Update an existing object.""" try: - instance = await get_object_or_404_async(model, id=item_id) + item_id_value = parse_model_id(model, item_id) + instance = await get_object_or_404_async(model, id=item_id_value) if before_update: payload = await execute_hook_async(before_update, request, instance, payload, update_schema) or payload @@ -341,10 +343,11 @@ async def update_item(request, item_id: int, payload: update_schema = Form(...)) return await handle_exception_async(e) else: @router.patch("/{item_id}", response=detail_schema, tags=[model.__name__], operation_id=f"update_{model_name}") - async def update_item(request, item_id: int, payload: update_schema) -> Any: #type: ignore + async def update_item(request, item_id: str, payload: update_schema) -> Any: #type: ignore """Update an existing object.""" try: - instance = await get_object_or_404_async(model, id=item_id) + item_id_value = parse_model_id(model, item_id) + instance = await get_object_or_404_async(model, id=item_id_value) if before_update: payload = await execute_hook_async(before_update, request, instance, payload, update_schema) or payload @@ -365,10 +368,11 @@ async def update_item(request, item_id: int, payload: update_schema) -> Any: #ty return await handle_exception_async(e) @router.delete("/{item_id}", response={200: Dict[str, str]}, tags=[model.__name__], operation_id=f"delete_{model_name}") - async def delete_item(request, item_id: int) -> Dict[str, str]: + async def delete_item(request, item_id: str) -> Dict[str, str]: """Delete an object.""" try: - instance = await get_object_or_404_async(model, id=item_id) + item_id_value = parse_model_id(model, item_id) + instance = await get_object_or_404_async(model, id=item_id_value) if before_delete: await execute_hook_async(before_delete, request, instance) @@ -399,10 +403,11 @@ def list_items(request, q: Optional[str] = None, sort: Optional[str] = None, return handle_exception(e) @router.get("/{item_id}", response=detail_schema, tags=[model.__name__], operation_id=f"get_{model_name}") - def get_item(request, item_id: int) -> Any: + def get_item(request, item_id: str) -> Any: """Retrieve a single object by ID.""" try: - instance = get_object_or_404(model, id=item_id) + item_id_value = parse_model_id(model, item_id) + instance = get_object_or_404(model, id=item_id_value) return handle_response(instance, detail_schema, custom_response, request) except Exception as e: return handle_exception(e) @@ -508,10 +513,11 @@ def create_item(request, payload: create_schema) -> Any: #type: ignore if update_schema: if use_multipart_update: @router.patch("/{item_id}", response=detail_schema, tags=[model.__name__], operation_id=f"update_{model_name}") - def update_item(request, item_id: int, payload: update_schema = Form(...)) -> Any: #type: ignore + def update_item(request, item_id: str, payload: update_schema = Form(...)) -> Any: #type: ignore """Update an existing object.""" try: - instance = get_object_or_404(model, id=item_id) + item_id_value = parse_model_id(model, item_id) + instance = get_object_or_404(model, id=item_id_value) if before_update: payload = execute_hook(before_update, request, instance, payload, update_schema) or payload @@ -593,10 +599,11 @@ def update_item(request, item_id: int, payload: update_schema = Form(...)) -> An else: @router.patch("/{item_id}", response=detail_schema, tags=[model.__name__], operation_id=f"update_{model_name}") - def update_item(request, item_id: int, payload: update_schema) -> Any: #type: ignore + def update_item(request, item_id: str, payload: update_schema) -> Any: #type: ignore """Update an existing object.""" try: - instance = get_object_or_404(model, id=item_id) + item_id_value = parse_model_id(model, item_id) + instance = get_object_or_404(model, id=item_id_value) if before_update: payload = execute_hook(before_update, request, instance, payload, update_schema) or payload data = convert_foreign_keys(model, payload.model_dump(exclude_unset=True)) @@ -612,10 +619,11 @@ def update_item(request, item_id: int, payload: update_schema) -> Any: #type: ig return handle_exception(e) @router.delete("/{item_id}", response={200: Dict[str, str]}, tags=[model.__name__], operation_id=f"delete_{model_name}") - def delete_item(request, item_id: int) -> Dict[str, str]: + def delete_item(request, item_id: str) -> Dict[str, str]: """Delete an object.""" try: - instance = get_object_or_404(model, id=item_id) + item_id_value = parse_model_id(model, item_id) + instance = get_object_or_404(model, id=item_id_value) if before_delete: execute_hook(before_delete, request, instance) instance.delete() diff --git a/src/lazy_ninja/utils.py b/src/lazy_ninja/utils.py deleted file mode 100644 index 67f7c3d..0000000 --- a/src/lazy_ninja/utils.py +++ /dev/null @@ -1,169 +0,0 @@ -from typing import Type, List, Optional -from pydantic import ConfigDict, create_model, model_validator -from decimal import Decimal -import asyncio - -from django.db import models -from django.shortcuts import get_object_or_404 -from asgiref.sync import sync_to_async - -from ninja import Schema - -from .helpers import execute_hook, apply_filters - -# Async version of general utils -convert_foreign_keys_async = sync_to_async(lambda model, data: convert_foreign_keys(model, data)) -get_all_objects_async = sync_to_async(lambda m: m.objects.all()) -get_object_or_404_async = sync_to_async(get_object_or_404) -execute_hook_async = sync_to_async(execute_hook) - -def convert_foreign_keys(model, data: dict) -> dict: - """ - Converts integer values for ForeignKey fields in `data` to the corresponding model instances. - """ - for field in model._meta.fields: # pylint: disable=W0212 - if isinstance(field, models.ForeignKey) and field.name in data: - fk_value = data[field.name] - if isinstance(fk_value, int): - # Retrieve the related model instance using the primary key. - data[field.name] = field.related_model.objects.get(pk=fk_value) - return data - -def is_async_context(): - """Check if we're in an async context by inspecting the call stack""" - try: - return asyncio.get_event_loop().is_running() - except RuntimeError: - return False - -def get_field_value_safely(obj, field): - """ - Get field value without triggering database queries. - For ForeignKey fields, returns the ID directly. - """ - if isinstance(field, models.ForeignKey): - attname = field.attname - if hasattr(obj, attname): - return getattr(obj, attname) - - return None - - # For non-ForeignKey fields, get the value directly - try: - value = getattr(obj, field.name) - return value - except Exception: - return None - -def serialize_model_instance(obj): - """ - Serializes a Django model instance into a dictionary with simple types. - Avoids triggering database queries for related fields. - """ - data = {} - for field in obj._meta.fields: - try: - value = get_field_value_safely(obj, field) - - if value is None: - data[field.name] = None - elif isinstance(field, (models.DateField, models.DateTimeField)): - data[field.name] = value.isoformat() if value else None - elif isinstance(field, (models.ImageField, models.FileField)): - data[field.name] = value.url if hasattr(value, 'url') else str(value) - elif hasattr(value, 'pk'): - data[field.name] = value.pk - else: - data[field.name] = value - except Exception: - # If we can't access a field, just set it to None - data[field.name] = None - return data - -serialize_model_instance_async = sync_to_async(serialize_model_instance) - -async def handle_response_async(instance, schema, custom_response, request): - """ - Async version of handle_response that uses serialize_model_instance_async - """ - if custom_response: - return await sync_to_async(custom_response)(request, instance) - - serialized = await serialize_model_instance_async(instance) - return serialized - -def get_pydantic_type(field) -> Type: - """ - Map a Django model field to an equivalent Python type for Pydantic validation. - """ - if isinstance(field, models.AutoField): - return int - elif isinstance(field, (models.CharField, models.TextField)): - return str - elif isinstance(field, models.IntegerField): - return int - elif isinstance(field, models.DecimalField): - return Decimal - elif isinstance(field, models.FloatField): - return float - elif isinstance(field, models.BooleanField): - return bool - elif isinstance(field, (models.DateField, models.DateTimeField)): - return str - elif isinstance(field, (models.ImageField, models.FileField)): - return str - elif isinstance(field, models.ForeignKey): - return int - else: - return str - -def generate_schema(model, exclude: List[str] = [], optional_fields: List[str] = [], update: bool = False) -> Type[Schema]: - """ - Generate a Pydantic schema based on a Django model. - - Parameters: - - model: The Django model class. - - exclude: A list of field names to exclude from the schema. - - optional_fields: A list of field names that should be marked as optional. - - Returns: - - A dynamically created Pydantic schema class. - - Notes: - - Fields listed in `optional_fields` or with null=True in the Django model are set as Optional. - - A root validator is added to preprocess the input using `serialize_model_instance`. - """ - fields = {} - for field in model._meta.fields: - if field.name in exclude: - continue - pydantic_type = get_pydantic_type(field) - - if update: - fields[field.name] = (Optional[pydantic_type], None) - - # Mark field as optional if it's in optional_fields or if the Django field allows null values. - elif field.name in optional_fields or field.null: - fields[field.name] = (Optional[pydantic_type], None) - else: - fields[field.name] = (pydantic_type, ...) - - class DynamicSchema(Schema): - @model_validator(mode="before") - def pre_serialize(cls, values): - """Define a pre-root validator that converts a Django model instance into a dict - using our serialize_model_instance function. - """ - if hasattr(values, "_meta"): - return serialize_model_instance(values) - return values - - model_config = ConfigDict(form_attributes=True) - - schema = create_model( - model.__name__ + "Schema", - __base__=DynamicSchema, - **fields - ) - - return schema diff --git a/src/lazy_ninja/utils/base.py b/src/lazy_ninja/utils/base.py index fe16807..2ab01f2 100644 --- a/src/lazy_ninja/utils/base.py +++ b/src/lazy_ninja/utils/base.py @@ -11,6 +11,7 @@ from django.shortcuts import get_object_or_404 + def convert_foreign_keys(model: Type[models.Model], data: Dict[str, Any]) -> Dict[str, Any]: """ Converts integer values for ForeignKey fields in `data` to the corresponding model instances. @@ -25,7 +26,7 @@ def convert_foreign_keys(model: Type[models.Model], data: Dict[str, Any]) -> Dic for field in model._meta.fields: if isinstance(field, models.ForeignKey) and field.name in data: fk_value = data[field.name] - if isinstance(fk_value, int): + if isinstance(fk_value, (int, str)): data[field.name] = field.related_model.objects.get(pk=fk_value) return data @@ -74,10 +75,18 @@ def serialize_model_instance(obj: models.Model) -> Dict[str, Any]: if value is None: data[field.name] = None + elif isinstance(field, models.UUIDField): + data[field.name] = str(value) elif isinstance(field, (models.DateField, models.DateTimeField)): data[field.name] = value.isoformat() if value else None elif isinstance(field, (models.ImageField, models.FileField)): data[field.name] = value.url if hasattr(value, 'url') else str(value) + elif isinstance(field, models.ForeignKey): + target_field = field.target_field + if isinstance(target_field, models.UUIDField): + data[field.name] = str(value) if value else None + else: + data[field.name] = value elif hasattr(value, 'pk'): data[field.name] = value.pk else: @@ -110,7 +119,9 @@ def get_pydantic_type(field: models.Field) -> Type: Returns: Python type for Pydantic """ - if isinstance(field, models.AutoField): + if isinstance(field, models.UUIDField): + return str + elif isinstance(field, models.AutoField): return int elif isinstance(field, (models.CharField, models.TextField)): return str @@ -127,7 +138,8 @@ def get_pydantic_type(field: models.Field) -> Type: elif isinstance(field, (models.ImageField, models.FileField)): return str elif isinstance(field, models.ForeignKey): - return int + target_field = field.target_field + return get_pydantic_type(target_field) else: return str diff --git a/src/lazy_ninja/utils/model.py b/src/lazy_ninja/utils/model.py index 90b7e4e..1d505b2 100644 --- a/src/lazy_ninja/utils/model.py +++ b/src/lazy_ninja/utils/model.py @@ -6,7 +6,6 @@ from .base import serialize_model_instance, serialize_model_instance_async - class BaseModelUtils: """Base class for model utilities."""