Saving my work
Drew Bednar
5 years ago
parent
59a05ca503
commit
262d32fea4
@ -0,0 +1,76 @@
|
||||
from inspect import Parameter
|
||||
from typing import List
|
||||
from molten import BaseApp, HTTPError, HTTP_409, HTTP_404
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from casbin_api.manager import BaseManager
|
||||
from casbin_api.error import EntityNotFound
|
||||
from .model import User, UserModel
|
||||
|
||||
|
||||
class UserManager(BaseManager):
|
||||
"""A `TodoManager` is accountable for the CRUD operations associated with a `Todo` instance"""
|
||||
|
||||
def schema_from_model(self, result: UserModel) -> User:
|
||||
_user = User(
|
||||
id=result.id,
|
||||
href=self.app.reverse_uri("get_user_by_id", user_id=result.id),
|
||||
createdDate=result.created_date,
|
||||
modifiedDate=result.modified_date,
|
||||
todo=result.todo,
|
||||
)
|
||||
return _user
|
||||
|
||||
def model_from_schema(self, user: User) -> UserModel:
|
||||
_user_model = UserModel(user.email, user.passwd)
|
||||
return _user_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,41 @@
|
||||
import bcrypt
|
||||
from typing import Optional
|
||||
from molten import schema, field
|
||||
from sqlalchemy import relationship, Column, Text, String, Boolean
|
||||
from casbin_api.db import Base, DBMixin
|
||||
from casbin_api.schema import Link
|
||||
|
||||
BCRYPT_ROUNDS = 11
|
||||
|
||||
|
||||
@schema
|
||||
class AuthToken:
|
||||
token: str = field(response_only=True)
|
||||
|
||||
|
||||
@schema
|
||||
class User:
|
||||
id: int = field(response_only=True)
|
||||
createdDate: str = field(response_only=True)
|
||||
modifiedDate: str = field(response_only=True)
|
||||
email: str
|
||||
passwd: str = field(request_only=True)
|
||||
href: Link = field(response_only=True)
|
||||
|
||||
|
||||
class UserModel(Base, DBMixin):
|
||||
__tablename__ = "user_account"
|
||||
email = Column(Text)
|
||||
passwd = Column(String(255))
|
||||
todos = relationship("TodoModel", back_populates="user")
|
||||
is_active = Column(Boolean, default=True)
|
||||
is_admin = Column(Boolean, default=False)
|
||||
|
||||
def __init__(self, email, passwd, is_admin=False):
|
||||
self.email = email
|
||||
self.passwd = bcrypt.hashpw(passwd.encode('utf-8', bcrypt.gensalt(BCRYPT_ROUNDS))).decode('utf-8')
|
||||
self.is_admin = is_admin
|
||||
|
||||
def check_password(self, passwd):
|
||||
"""Checks provided password against saved password hash."""
|
||||
return bcrypt.checkpw(passwd.encode('utf-8'), self.passwd.encode('utf-8'))
|
||||
@ -0,0 +1,41 @@
|
||||
from typing import List
|
||||
from molten import Route, Include, HTTP_201, HTTP_202, HTTPError, HTTP_404
|
||||
|
||||
from casbin_api.schema import APIResponse
|
||||
from casbin_api.error import EntityNotFound
|
||||
from .model import User, AuthToken
|
||||
from .manager import UserManager
|
||||
|
||||
|
||||
def create_user(user: User, user_manager: UserManager) -> User:
|
||||
_user = user_manager.create_user(user)
|
||||
headers = {"Location": _user.href}
|
||||
return HTTP_201, _user, headers
|
||||
|
||||
|
||||
def get_user_by_id(user_id: int, user_manager: UserManager) -> User:
|
||||
try:
|
||||
_user = user_manager.get_user_by_id(user_id)
|
||||
except EntityNotFound as err:
|
||||
raise HTTPError(HTTP_404, APIResponse(status=404, message=err.message))
|
||||
return _user
|
||||
|
||||
|
||||
# Update user only if you are the user or you are admin
|
||||
def update_user(user_id: int, user: User, user_manager: UserManager) -> User:
|
||||
return user_manager.update_todo(user_id, user)
|
||||
|
||||
|
||||
def login_user(user: User, user_manager: UserManager) -> AuthToken:
|
||||
pass
|
||||
|
||||
|
||||
auth_routes = Include(
|
||||
"/auth",
|
||||
[
|
||||
Route("/login", login_user, method="POST"),
|
||||
Route("/register", create_user, method="POST"),
|
||||
Route("/profile/{user_id}", get_user_by_id, method="GET"),
|
||||
Route("/profile/{user_id}", update_user, method="PUT"),
|
||||
],
|
||||
)
|
||||
@ -0,0 +1,57 @@
|
||||
def test_token_route(client):
|
||||
message = "welcome to casbin_api"
|
||||
response = client.get("/")
|
||||
content = response.json()
|
||||
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…
Reference in New Issue