Merge branch 'develop'
commit
4a505318d5
@ -0,0 +1,42 @@
|
||||
import pytest
|
||||
|
||||
from molten import testing
|
||||
from molten.contrib.sqlalchemy import EngineData
|
||||
|
||||
from {{cookiecutter.project_slug}}.index import create_app
|
||||
from {{cookiecutter.project_slug}}.db import Base
|
||||
|
||||
|
||||
|
||||
# requires function scope so that database is removed on every tests
|
||||
@pytest.fixture(scope="function")
|
||||
def app():
|
||||
app = create_app()
|
||||
yield app
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def create_db(app):
|
||||
"""Creates a test database with session scope"""
|
||||
def _retrieve_engine(engine_data: EngineData):
|
||||
return engine_data.engine
|
||||
|
||||
engine = app.injector.get_resolver().resolve(_retrieve_engine)()
|
||||
|
||||
Base.metadata.create_all(bind=engine)
|
||||
|
||||
yield
|
||||
|
||||
Base.metadata.drop_all(bind=engine)
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def client(app):
|
||||
"""Creates a testing client"""
|
||||
return testing.TestClient(app)
|
||||
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def session():
|
||||
pass
|
||||
@ -1,12 +1,39 @@
|
||||
from molten.testing import TestClient
|
||||
from {{cookiecutter.project_slug}}.index import create_app
|
||||
def test_welcome_route(client):
|
||||
message = "welcome to {{cookiecutter.project_slug}}"
|
||||
response = client.get("/")
|
||||
content = response.json()
|
||||
assert message == content.get("message")
|
||||
|
||||
app = create_app()
|
||||
test_client = TestClient(app)
|
||||
|
||||
def test_empty_get_todos(client):
|
||||
response = client.get("/todos")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == []
|
||||
|
||||
def test_welcome_route():
|
||||
message = "welcome to {{cookiecutter.project_slug}}"
|
||||
response = test_client.get("/")
|
||||
|
||||
def test_insert_todo(client):
|
||||
payload = {"todo": "walk the dog"}
|
||||
response = client.post("/todos", data=payload)
|
||||
content = response.json()
|
||||
assert message == content.get("message")
|
||||
assert response.status_code == 201
|
||||
assert type(content['id']) == int
|
||||
assert content['todo'] == payload['todo']
|
||||
|
||||
|
||||
def test_update_todo(client):
|
||||
payload = {"todo": "sample app"}
|
||||
response = client.post("/todos", json=payload)
|
||||
todo = response.json()
|
||||
print(todo)
|
||||
update_response = client.patch("{}".format(todo.get("href")), json={"complete": True, "todo": "sample app"})
|
||||
updated_todo = update_response.json()
|
||||
print(updated_todo)
|
||||
assert updated_todo["complete"] == True
|
||||
|
||||
|
||||
def test_todo_not_found(client):
|
||||
response = client.get("/todo/1111111")
|
||||
content = response.json()
|
||||
assert response.status_code == 404
|
||||
assert content["status"] == 404
|
||||
assert content["message"]
|
||||
|
||||
@ -0,0 +1,2 @@
|
||||
from .manager import TodoManager, TodoManagerComponent
|
||||
from .view import todo_routes
|
||||
@ -0,0 +1,80 @@
|
||||
from inspect import Parameter
|
||||
from typing import List
|
||||
from molten import BaseApp, HTTPError, HTTP_409, HTTP_404
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from {{cookiecutter.project_slug}}.manager import BaseManager
|
||||
from {{cookiecutter.project_slug}}.error import EntityNotFound
|
||||
from .model import Todo, TodoModel
|
||||
|
||||
|
||||
class TodoManager(BaseManager):
|
||||
"""A `TodoManager` is accountable for the CRUD operations associated with a `Todo` instance"""
|
||||
|
||||
def schema_from_model(self, result: TodoModel) -> Todo:
|
||||
_todo = Todo(
|
||||
id=result.id,
|
||||
href=self.app.reverse_uri("get_todo_by_id", todo_id=result.id),
|
||||
createdDate=result.created_date,
|
||||
modifiedDate=result.modified_date,
|
||||
todo=result.todo,
|
||||
complete=result.complete
|
||||
)
|
||||
return _todo
|
||||
|
||||
def model_from_schema(self, todo: Todo) -> TodoModel:
|
||||
_todo_model = TodoModel(
|
||||
todo=todo.todo,
|
||||
complete=todo.complete
|
||||
)
|
||||
return _todo_model
|
||||
|
||||
def get_todos(self) -> List[Todo]:
|
||||
"""Retrieves a list of `Todo` representations"""
|
||||
results = self.session.query(TodoModel).order_by(TodoModel.id).all()
|
||||
todos = [self.schema_from_model(result) for result in results]
|
||||
return todos
|
||||
|
||||
def get_todo_by_id(self, id) -> Todo:
|
||||
"""Retrieves a `Todo` representation by id"""
|
||||
result = self.session.query(TodoModel).filter_by(id=id).one_or_none()
|
||||
if result is None:
|
||||
raise EntityNotFound(f"Todo: {id} does not exist")
|
||||
return self.schema_from_model(result)
|
||||
|
||||
def create_todo(self, todo: Todo) -> Todo:
|
||||
"""Creates a new `Todo` resource and returns its representation"""
|
||||
todo_model = self.model_from_schema(todo)
|
||||
self.session.add(todo_model)
|
||||
self.session.flush()
|
||||
return self.schema_from_model(todo_model)
|
||||
|
||||
def update_todo(self, todo_id: int, todo: Todo) -> Todo:
|
||||
"""Updates an existing `Todo` resource and returns its new representation"""
|
||||
result = self.session.query(TodoModel).filter_by(id=todo_id).one_or_none()
|
||||
if result is None:
|
||||
raise EntityNotFound(f"Todo: {todo_id} does not exist")
|
||||
updates = self.model_from_schema(todo)
|
||||
updates.id = todo_id
|
||||
self.session.merge(updates)
|
||||
self.session.flush()
|
||||
todo = self.schema_from_model(result)
|
||||
return todo
|
||||
|
||||
def delete_todo(self, id):
|
||||
"""Deletes a `Todo` """
|
||||
result = self.session.query(TodoModel).filter_by(id=id).one_or_none()
|
||||
if result is not None:
|
||||
self.session.delete(result)
|
||||
return
|
||||
|
||||
|
||||
class TodoManagerComponent:
|
||||
is_cacheable = True
|
||||
is_singleton = False
|
||||
|
||||
def can_handle_parameter(self, parameter: Parameter) -> bool:
|
||||
return parameter.annotation is TodoManager
|
||||
|
||||
def resolve(self, session: Session, app: BaseApp) -> TodoManager: # type: ignore
|
||||
return TodoManager(session, app)
|
||||
@ -0,0 +1,21 @@
|
||||
from typing import Optional
|
||||
from molten import schema, field
|
||||
from sqlalchemy import Column, Text, Boolean
|
||||
from {{cookiecutter.project_slug}}.db import Base, DBMixin
|
||||
from {{cookiecutter.project_slug}}.schema import Link
|
||||
|
||||
|
||||
@schema
|
||||
class Todo:
|
||||
id: int = field(response_only=True)
|
||||
createdDate: str = field(response_only=True)
|
||||
modifiedDate: str = field(response_only=True)
|
||||
todo: str
|
||||
complete: Optional[bool]
|
||||
href: Link = field(response_only=True)
|
||||
|
||||
|
||||
class TodoModel(Base, DBMixin):
|
||||
__tablename__ = "todo"
|
||||
todo = Column(Text)
|
||||
complete = Column(Boolean, default=False)
|
||||
@ -0,0 +1,49 @@
|
||||
from typing import List
|
||||
from molten import Route, Include, HTTP_201, HTTP_202, HTTPError, HTTP_404
|
||||
|
||||
from {{cookiecutter.project_slug}}.schema import APIResponse
|
||||
from {{cookiecutter.project_slug}}.error import EntityNotFound
|
||||
from .model import Todo
|
||||
from .manager import TodoManager
|
||||
|
||||
|
||||
def list_todos(todo_manager: TodoManager) -> List[Todo]:
|
||||
return todo_manager.get_todos()
|
||||
|
||||
|
||||
def create_todo(todo: Todo, todo_manager: TodoManager) -> Todo:
|
||||
_todo = todo_manager.create_todo(todo)
|
||||
headers = {"Location": _todo.href}
|
||||
return HTTP_201, _todo, headers
|
||||
|
||||
|
||||
def delete_todo(todo_id: int, todo_manager: TodoManager):
|
||||
todo_manager.delete_todo(todo_id)
|
||||
return (
|
||||
HTTP_202,
|
||||
APIResponse(status=202, message=f"Delete request for store: {todo_id} accepted"),
|
||||
)
|
||||
|
||||
|
||||
def get_todo_by_id(todo_id: int, todo_manager: TodoManager) -> Todo:
|
||||
try:
|
||||
_todo = todo_manager.get_todo_by_id(todo_id)
|
||||
except EntityNotFound as err:
|
||||
raise HTTPError(HTTP_404,
|
||||
APIResponse(status=404,
|
||||
message=err.message)
|
||||
)
|
||||
return _todo
|
||||
|
||||
|
||||
def update_todo(todo_id: int, todo: Todo, todo_manager: TodoManager) -> Todo:
|
||||
return todo_manager.update_todo(todo_id, todo)
|
||||
|
||||
|
||||
todo_routes = Include("/todos", [
|
||||
Route("", list_todos, method="GET"),
|
||||
Route("", create_todo, method="POST"),
|
||||
Route("/{todo_id}", delete_todo, method="DELETE"),
|
||||
Route("/{todo_id}", get_todo_by_id, method="GET"),
|
||||
Route("/{todo_id}", update_todo, method="PATCH")
|
||||
])
|
||||
@ -0,0 +1,32 @@
|
||||
from sqlalchemy import Column, BigInteger, DateTime
|
||||
from sqlalchemy.ext.compiler import compiles
|
||||
from sqlalchemy.ext.declarative import declarative_base
|
||||
from sqlalchemy.schema import CreateColumn
|
||||
from sqlalchemy.sql import expression
|
||||
from sqlalchemy.types import DateTime as DatetimeType
|
||||
|
||||
Base = declarative_base()
|
||||
|
||||
|
||||
class utcnow(expression.FunctionElement):
|
||||
type = DatetimeType()
|
||||
|
||||
|
||||
@compiles(utcnow, "postgresql")
|
||||
def pg_utcnow(element, compiler, **kw):
|
||||
return "TIMEZONE('utc', CURRENT_TIMESTAMP)"
|
||||
|
||||
|
||||
@compiles(CreateColumn, 'postgresql')
|
||||
def use_identity(element, compiler, **kw):
|
||||
text = compiler.visit_create_column(element, **kw)
|
||||
text = text.replace("SERIAL", "INT GENERATED BY DEFAULT AS IDENTITY")
|
||||
return text
|
||||
|
||||
|
||||
class DBMixin:
|
||||
id = Column(BigInteger, primary_key=True)
|
||||
created_date = Column(DateTime(timezone=True), server_default=utcnow())
|
||||
modified_date = Column(
|
||||
DateTime(timezone=True), server_default=utcnow(), onupdate=utcnow()
|
||||
)
|
||||
@ -0,0 +1,5 @@
|
||||
from molten.errors import MoltenError
|
||||
|
||||
|
||||
class EntityNotFound(MoltenError):
|
||||
"""Raised when an entity is not found using an `exists` check in sqlalchemy."""
|
||||
@ -0,0 +1,32 @@
|
||||
from abc import ABCMeta, abstractmethod
|
||||
from molten import BaseApp, HTTPError, HTTP_409, HTTP_404
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
|
||||
class BaseManager(metaclass=ABCMeta):
|
||||
"""Base instance for Model managers"""
|
||||
|
||||
def __init__(self, session: Session, app: BaseApp):
|
||||
self.session = session
|
||||
self.app = app
|
||||
|
||||
@abstractmethod
|
||||
def model_from_schema(self, schema):
|
||||
"""Converts a Schema instance into a SQLAlchemy ORM model instance"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def schema_from_model(self, result):
|
||||
"""Converts a SQLAlchemy results proxy into a Schema instance"""
|
||||
pass
|
||||
|
||||
def raise_409(self, id:int):
|
||||
"""Raises a 409 HTTP error response in the event of Conflict"""
|
||||
raise HTTPError(
|
||||
HTTP_409,
|
||||
{
|
||||
"status": 409,
|
||||
"message": f"Entity {self.__class__.__name__} with id: {id} already exists"
|
||||
}
|
||||
|
||||
)
|
||||
Loading…
Reference in New Issue