Removed example Todo endpoints and tests that the cookiecutter provided

pull/1/head
androiddrew 6 years ago
parent 50014ae1ff
commit e32d5908a0

@ -1,2 +0,0 @@
from .manager import TodoManager, TodoManagerComponent
from .view import todo_routes

@ -1,80 +0,0 @@
from inspect import Parameter
from typing import List
from molten import BaseApp, HTTPError, HTTP_409, HTTP_404
from sqlalchemy.orm import Session
from pulley.manager import BaseManager
from pulley.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)

@ -1,21 +0,0 @@
from typing import Optional
from molten import schema, field
from sqlalchemy import Column, Text, Boolean
from pulley.db import Base, DBMixin
from pulley.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)

@ -1,49 +0,0 @@
from typing import List
from molten import Route, Include, HTTP_201, HTTP_202, HTTPError, HTTP_404
from pulley.schema import APIResponse
from pulley.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 todo: {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")
])

@ -8,7 +8,6 @@ from wsgicors import CORS
from whitenoise import WhiteNoise from whitenoise import WhiteNoise
from .api.welcome import welcome from .api.welcome import welcome
# from .api.todo import TodoManagerComponent, todo_routes
from .common import ExtJSONRenderer from .common import ExtJSONRenderer
from .logger import setup_logging from .logger import setup_logging
from .schema import APIResponse from .schema import APIResponse
@ -28,7 +27,6 @@ components = [
SettingsComponent(settings), SettingsComponent(settings),
SQLAlchemyEngineComponent(), SQLAlchemyEngineComponent(),
SQLAlchemySessionComponent(), SQLAlchemySessionComponent(),
# TodoManagerComponent(),
] ]
middleware = [ResponseRendererMiddleware(), SQLAlchemyMiddleware()] middleware = [ResponseRendererMiddleware(), SQLAlchemyMiddleware()]
@ -39,7 +37,7 @@ routes = [
Route("/", welcome, "GET"), Route("/", welcome, "GET"),
Route("/_schema", get_schema, "GET"), Route("/_schema", get_schema, "GET"),
Route("/_docs", get_docs, "GET"), Route("/_docs", get_docs, "GET"),
] # + [todo_routes] ]
class ExtApp(App): class ExtApp(App):

@ -3,53 +3,3 @@ def test_welcome_route(client):
response = client.get("/") response = client.get("/")
content = response.json() content = response.json()
assert message == content.get("message") assert message == content.get("message")
#
#
# def test_empty_get_todos(client):
# response = client.get("/todos")
# assert response.status_code == 200
# assert response.json() == []
#
#
# def test_insert_todo(client):
# payload = {"todo": "walk the dog"}
# response = client.post("/todos", data=payload)
# content = response.json()
# assert response.status_code == 201
# assert type(content['id']) == int
# assert content['todo'] == payload['todo']
#
#
# def test_get_individual_todo_by_href(client):
# payload = {"todo": "my individual todo"}
# response = client.post("/todos", data=payload)
# content = response.json()
# get_response = client.get(f"{content.get('href')}")
# get_content = get_response.json()
# assert get_response.status_code == 200
# assert content == get_content
#
#
# def test_update_todo(client):
# payload = {"todo": "sample app"}
# response = client.post("/todos", json=payload)
# todo = response.json()
# update_response = client.patch("{}".format(todo.get("href")), json={"complete": True, "todo": "sample app"})
# updated_todo = update_response.json()
# 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"]
#
#
# def test_delete_todo(client):
# payload = {"todo": "sample app"}
# response = client.post("/todos", json=payload)
# todo = response.json()
# delete_response = client.delete(f"/todos/{todo.get('id')}")
# assert delete_response.status_code == 202

Loading…
Cancel
Save