Skip to content

Commit

Permalink
Merge pull request #160 from thomaxxl/annot_form
Browse files Browse the repository at this point in the history
Annotation & formatting
  • Loading branch information
thomaxxl authored Sep 5, 2024
2 parents 186b9b5 + 53ead50 commit fbcd493
Show file tree
Hide file tree
Showing 6 changed files with 219 additions and 47 deletions.
3 changes: 2 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,8 @@ classifiers = [ # Optional

# Indicate who your project is intended for
"Intended Audience :: Developers",
"Topic :: Software Development :: API Development",
"Topic :: Software Development :: Libraries",
"Framework :: Flask",
"License :: OSI Approved :: GNU General Public License v3 (GPLv3)",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.7",
Expand Down
43 changes: 23 additions & 20 deletions safrs/_safrs_relationship.py
Original file line number Diff line number Diff line change
@@ -1,34 +1,37 @@
from http import HTTPStatus
from typing import Dict, Tuple, Any
from .util import classproperty


# pylint: disable=too-few-public-methods
class SAFRSRelationshipObject:
"""
Relationship object, used to emulate a SAFRSBase object for the swagger for relationship targets
so we can call the same methods on a relationship target as we do when using SAFRSBase
so we can call the same methods on a relationship target as we do when using SAFRSBase.
"""

_s_class_name = None
__name__ = "name"
http_methods = {"GET", "POST", "PATCH", "DELETE"}
swagger_models = {"instance": None, "collection": None}
_s_class_name: str = None
__name__: str = "name"
http_methods: set = {"GET", "POST", "PATCH", "DELETE"}
swagger_models: Dict[str, Any] = {"instance": None, "collection": None}

@classmethod
def _s_get_swagger_doc(cls, http_method):
"""Create a swagger api model based on the sqlalchemy schema
if an instance exists in the DB, the first entry is used as example
def _s_get_swagger_doc(cls, http_method: str) -> Tuple[Dict[str, Any], Dict[str, Any]]:
"""
Create a swagger API model based on the SQLAlchemy schema.
If an instance exists in the DB, the first entry is used as an example.
:param http_method: HTTP method for which to generate the doc
:return: swagger body, responses
:return: Tuple containing the swagger body and responses
"""
body = {}
responses = {}
object_name = cls.__name__
body: Dict[str, Any] = {}
responses: Dict[str, Any] = {}
object_name: str = cls.__name__

object_model = {}
object_model: Dict[str, Any] = {}
responses = {str(HTTPStatus.OK.value): {"description": f"{object_name} object", "schema": object_model}}

if http_method.upper() in ("POST", "GET"):
if http_method.upper() in {"POST", "GET"}:
responses = {
str(HTTPStatus.OK.value): {"description": HTTPStatus.OK.description},
str(HTTPStatus.NOT_FOUND.value): {"description": HTTPStatus.NOT_FOUND.description},
Expand All @@ -37,29 +40,29 @@ def _s_get_swagger_doc(cls, http_method):
return body, responses

@classproperty
def _s_relationships(cls):
def _s_relationships(cls) -> Any:
"""
:return: The relationship names of the target
"""
return cls._target._s_relationships

@classproperty
def _s_jsonapi_attrs(cls):
def _s_jsonapi_attrs(cls) -> Any:
"""
:return: target JSON:API attributes
:return: Target JSON:API attributes
"""
return cls._target._s_jsonapi_attrs

@classproperty
def _s_type(cls):
def _s_type(cls) -> str:
"""
:return: JSON:API type
"""
return cls._target._s_type

@classproperty
def _s_class_name(cls):
def _s_class_name(cls) -> str:
"""
:return: name of the target class
:return: Name of the target class
"""
return cls._target.__name__
32 changes: 13 additions & 19 deletions safrs/api_methods.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,4 @@
#
# jsonapi_rpc methods that can be added to the exposed classes
#
from typing import Any, Dict
from sqlalchemy import or_
from sqlalchemy.orm.session import make_transient
import safrs
Expand All @@ -11,7 +9,7 @@


@jsonapi_rpc(http_methods=["POST"])
def duplicate(self):
def duplicate(self: Any) -> SAFRSFormattedResponse:
"""
description: Duplicate an object - copy it and give it a new id
"""
Expand All @@ -21,16 +19,15 @@ def duplicate(self):
self.id = self.id_type()
session.add(self)
session.commit()
response = SAFRSFormattedResponse(self)
return response
return SAFRSFormattedResponse(self)


@classmethod
@jsonapi_rpc(http_methods=["POST"])
def lookup_re_mysql(cls, **kwargs): # pragma: no cover
def lookup_re_mysql(cls: Any, **kwargs: Dict[str, str]) -> SAFRSFormattedResponse: # pragma: no cover
"""
pageable: True
description : Regex search all matching objects (works only in MySQL!!!)
description: Regex search all matching objects (works only in MySQL!!!)
args:
name: thom.*
"""
Expand All @@ -49,10 +46,10 @@ def lookup_re_mysql(cls, **kwargs): # pragma: no cover

@classmethod
@jsonapi_rpc(http_methods=["POST"])
def startswith(cls, **kwargs): # pragma: no cover
def startswith(cls: Any, **kwargs: Dict[str, str]) -> SAFRSFormattedResponse: # pragma: no cover
"""
pageable: True
summary : lookup items where specified attributes starts with the argument string
summary: Lookup items where specified attributes start with the argument string
args:
attr_name: value
"""
Expand All @@ -79,18 +76,17 @@ def startswith(cls, **kwargs): # pragma: no cover
meta = {}
errors = None
response = SAFRSFormattedResponse(data, meta, links, errors, count)

except Exception as exc:
raise GenericError(f"Failed to execute query {exc}")
return response


@classmethod
@jsonapi_rpc(http_methods=["POST"])
def search(cls, **kwargs): # pragma: no cover
def search(cls: Any, **kwargs: Dict[str, str]) -> SAFRSFormattedResponse: # pragma: no cover
"""
pageable: True
description : lookup column names
description: Lookup column names
args:
query: val
"""
Expand All @@ -106,16 +102,15 @@ def search(cls, **kwargs): # pragma: no cover
data = [item for item in instances]
meta = {}
errors = None
response = SAFRSFormattedResponse(data, meta, links, errors, count)
return response
return SAFRSFormattedResponse(data, meta, links, errors, count)


@classmethod
@jsonapi_rpc(http_methods=["POST"])
def re_search(cls, **kwargs): # pragma: no cover
def re_search(cls: Any, **kwargs: Dict[str, str]) -> SAFRSFormattedResponse: # pragma: no cover
"""
pageable: True
description : lookup column names
description: Lookup column names
args:
query: search.*all
"""
Expand All @@ -126,5 +121,4 @@ def re_search(cls, **kwargs): # pragma: no cover
data = [item for item in instances]
meta = {}
errors = None
response = SAFRSFormattedResponse(data, meta, links, errors, count)
return response
return SAFRSFormattedResponse(data, meta, links, errors, count)
174 changes: 174 additions & 0 deletions safrs/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,180 @@
#
# pylint: disable=logging-format-interpolation,no-self-argument,no-member,line-too-long,fixme,protected-access
#
"""
SAFRSBase class customizable attributes and methods, override these to customize the behavior of the SAFRSBase class.
http_methods:
Type: List[str]
A list of HTTP methods that are allowed for this class when exposed in the API.
Common methods include 'GET', 'POST', 'PUT', 'DELETE', etc.
This property controls the types of operations that can be performed on instances
of the class via the API.
_s_post:
Type: classmethod
Description: Called when a new item is created with a POST to the JSON:API.
_s_patch:
Type: method
Description: Updates the object attributes.
_s_delete:
Type: method
Description: Deletes the instance from the database.
_s_get:
Type: classmethod
Description: Called when a collection is requested with an HTTP GET to the JSON:API.
_s_expose:
Type: bool
Description: Indicates whether this class should be exposed in the API.
_s_upsert:
Type: bool
Description: Indicates whether to look up and use existing objects during creation.
_s_allow_add_rels:
Type: bool
Description: Allows relationships to be added in POST requests.
_s_pk_delimiter:
Type: str
Description: Delimiter used for primary keys.
_s_url_root:
Type: Optional[str]
Description: URL prefix shown in the "links" field. If not set, request.url_root will be used.
_s_columns:
Type: classproperty
Description: List of columns that are exposed by the API.
_s_relationships:
Type: hybrid_property
Description: Dictionary of relationships used for JSON:API (de)serialization.
_s_jsonapi_attrs:
Type: hybrid_property
Description: Dictionary of exposed attribute names and values.
_s_auto_commit:
Type: classproperty
Description: Indicates whether the instance should be automatically committed.
_s_check_perm:
Type: hybrid_method
Description: Checks the (instance-level) column permission.
_s_jsonapi_encode:
Type: hybrid_method
Description: Encodes the object according to the JSON:API specification.
_s_get_related:
Type: method
Description: Returns a dictionary of relationship names to related instances.
_s_count:
Type: classmethod
Description: Returns the count of instances in the table.
_s_sample_dict:
Type: classmethod
Description: Returns a sample dictionary to be used as an example "attributes" payload in the Swagger example.
_s_object_id:
Type: classproperty
Description: Returns the Flask URL parameter name of the object.
_s_get_jsonapi_rpc_methods:
Type: classmethod
Description: Returns a list of JSON:API RPC methods for this class.
_s_get_swagger_doc:
Type: classmethod
Description: Returns the Swagger body and response dictionaries for the specified HTTP method.
_s_sample_id:
Type: classmethod
Description: Returns a sample ID for the API documentation.
_s_url:
Type: hybrid_property
Description: Returns the endpoint URL of this instance.
_s_meta:
Type: classmethod
Description: Returns the "meta" part of the response.
_s_query:
Type: classproperty
Description: Returns the SQLAlchemy query object.
_s_class_name:
Type: classproperty
Description: Returns the name of the instances.
_s_collection_name:
Type: classproperty
Description: Returns the name of the collection, used to construct the endpoint.
_s_type:
Type: classproperty
Description: Returns the JSON:API "type", i.e., the table name if this is a DB model, the class name otherwise.
_s_expunge:
Type: method
Description: Expunges an object from its session.
_s_get_instance_by_id:
Type: classmethod
Description: Returns the query object for the specified JSON:API ID.
_s_parse_attr_value:
Type: method
Description: Parses the given JSON:API attribute value so it can be stored in the DB.
_s_clone:
Type: method
Description: Clones an object by copying the parameters and creating a new ID.
_s_filter:
Type: classmethod
Description: Applies filters to the query.
"""
from __future__ import annotations
import inspect
import datetime
Expand Down
Loading

0 comments on commit fbcd493

Please sign in to comment.