Add example of third party API
This commit is contained in:
2
app/other_api/doodad/__init__.py
Normal file
2
app/other_api/doodad/__init__.py
Normal file
@@ -0,0 +1,2 @@
|
||||
from .model import Doodad # noqa
|
||||
from .schema import DoodadSchema # noqa
|
56
app/other_api/doodad/controller.py
Normal file
56
app/other_api/doodad/controller.py
Normal file
@@ -0,0 +1,56 @@
|
||||
from flask import request
|
||||
from flask_accepts import accepts, responds
|
||||
from flask_restplus import Namespace, Resource
|
||||
from flask.wrappers import Response
|
||||
from typing import List
|
||||
|
||||
from .schema import DoodadSchema
|
||||
from .service import DoodadService
|
||||
from .model import Doodad
|
||||
from .interface import DoodadInterface
|
||||
|
||||
api = Namespace('Doodad', description='A modular namespace within Other API') # noqa
|
||||
|
||||
|
||||
@api.route('/')
|
||||
class DoodadResource(Resource):
|
||||
'''Doodads'''
|
||||
|
||||
@responds(schema=DoodadSchema, many=True)
|
||||
def get(self) -> List[Doodad]:
|
||||
'''Get all Doodads'''
|
||||
|
||||
return DoodadService.get_all()
|
||||
|
||||
@accepts(schema=DoodadSchema, api=api)
|
||||
@responds(schema=DoodadSchema)
|
||||
def post(self) -> Doodad:
|
||||
'''Create a Single Doodad'''
|
||||
|
||||
return DoodadService.create(request.parsed_obj)
|
||||
|
||||
|
||||
@api.route('/<int:doodadId>')
|
||||
@api.param('doodadId', 'Doodad database ID')
|
||||
class DoodadIdResource(Resource):
|
||||
@responds(schema=DoodadSchema)
|
||||
def get(self, doodadId: int) -> Doodad:
|
||||
'''Get Single Doodad'''
|
||||
|
||||
return DoodadService.get_by_id(doodadId)
|
||||
|
||||
def delete(self, doodadId: int) -> Response:
|
||||
'''Delete Single Doodad'''
|
||||
from flask import jsonify
|
||||
print('doodadId = ', doodadId)
|
||||
id = DoodadService.delete_by_id(doodadId)
|
||||
return jsonify(dict(status='Success', id=id))
|
||||
|
||||
@accepts(schema=DoodadSchema, api=api)
|
||||
@responds(schema=DoodadSchema)
|
||||
def put(self, doodadId: int) -> Doodad:
|
||||
'''Update Single Doodad'''
|
||||
|
||||
changes: DoodadInterface = request.parsed_obj
|
||||
Doodad = DoodadService.get_by_id(doodadId)
|
||||
return DoodadService.update(Doodad, changes)
|
84
app/other_api/doodad/controller_test.py
Normal file
84
app/other_api/doodad/controller_test.py
Normal file
@@ -0,0 +1,84 @@
|
||||
|
||||
from unittest.mock import patch
|
||||
from flask.testing import FlaskClient
|
||||
|
||||
from app.test.fixtures import client, app # noqa
|
||||
from .service import DoodadService
|
||||
from .schema import DoodadSchema
|
||||
from .model import Doodad
|
||||
from .interface import DoodadInterface
|
||||
from .. import BASE_ROUTE
|
||||
|
||||
|
||||
def make_doodad(id: int = 123, name: str = 'Test doodad',
|
||||
purpose: str = 'Test purpose') -> Doodad:
|
||||
return Doodad(
|
||||
doodad_id=id, name=name, purpose=purpose
|
||||
)
|
||||
|
||||
|
||||
class TestDoodadResource:
|
||||
@patch.object(DoodadService, 'get_all',
|
||||
lambda: [make_doodad(123, name='Test Doodad 1'),
|
||||
make_doodad(456, name='Test Doodad 2')])
|
||||
def test_get(self, client: FlaskClient): # noqa
|
||||
with client:
|
||||
results = client.get(f'/api/{BASE_ROUTE}/doodad',
|
||||
follow_redirects=True).get_json()
|
||||
expected = DoodadSchema(many=True).dump(
|
||||
[make_doodad(123, name='Test Doodad 1'),
|
||||
make_doodad(456, name='Test Doodad 2')]
|
||||
).data
|
||||
for r in results:
|
||||
assert r in expected
|
||||
|
||||
@patch.object(DoodadService, 'create',
|
||||
lambda create_request: Doodad(**create_request))
|
||||
def test_post(self, client: FlaskClient): # noqa
|
||||
with client:
|
||||
|
||||
payload = dict(name='Test doodad', purpose='Test purpose')
|
||||
result = client.post(f'/api/{BASE_ROUTE}/doodad/', json=payload).get_json()
|
||||
expected = DoodadSchema().dump(Doodad(
|
||||
name=payload['name'],
|
||||
purpose=payload['purpose'],
|
||||
)).data
|
||||
assert result == expected
|
||||
|
||||
|
||||
def fake_update(doodad: Doodad, changes: DoodadInterface) -> Doodad:
|
||||
# To fake an update, just return a new object
|
||||
updated_Doodad = Doodad(doodad_id=doodad.doodad_id,
|
||||
name=changes['name'],
|
||||
purpose=changes['purpose'])
|
||||
return updated_Doodad
|
||||
|
||||
|
||||
class TestDoodadIdResource:
|
||||
@patch.object(DoodadService, 'get_by_id',
|
||||
lambda id: make_doodad(id=id))
|
||||
def test_get(self, client: FlaskClient): # noqa
|
||||
with client:
|
||||
result = client.get(f'/api/{BASE_ROUTE}/doodad/123').get_json()
|
||||
expected = Doodad(doodad_id=123)
|
||||
assert result['doodadId'] == expected.doodad_id
|
||||
|
||||
@patch.object(DoodadService, 'delete_by_id',
|
||||
lambda id: [id])
|
||||
def test_delete(self, client: FlaskClient): # noqa
|
||||
with client:
|
||||
result = client.delete(f'/api/{BASE_ROUTE}/doodad/123').get_json()
|
||||
expected = dict(status='Success', id=[123])
|
||||
assert result == expected
|
||||
|
||||
@patch.object(DoodadService, 'get_by_id',
|
||||
lambda id: make_doodad(id=id))
|
||||
@patch.object(DoodadService, 'update', fake_update)
|
||||
def test_put(self, client: FlaskClient): # noqa
|
||||
with client:
|
||||
result = client.put(f'/api/{BASE_ROUTE}/doodad/123',
|
||||
json={'name': 'New Doodad',
|
||||
'purpose': 'New purpose'}).get_json()
|
||||
expected = DoodadSchema().dump(
|
||||
Doodad(doodad_id=123, name='New Doodad', purpose='New purpose')).data
|
||||
assert result == expected
|
7
app/other_api/doodad/interface.py
Normal file
7
app/other_api/doodad/interface.py
Normal file
@@ -0,0 +1,7 @@
|
||||
from mypy_extensions import TypedDict
|
||||
|
||||
|
||||
class DoodadInterface(TypedDict, total=False):
|
||||
doodad_id: int
|
||||
name: str
|
||||
purpose: str
|
19
app/other_api/doodad/model.py
Normal file
19
app/other_api/doodad/model.py
Normal file
@@ -0,0 +1,19 @@
|
||||
from sqlalchemy import Integer, Column, String
|
||||
from app import db # noqa
|
||||
from .interface import DoodadInterface
|
||||
from typing import Any
|
||||
|
||||
|
||||
class Doodad(db.Model): # type: ignore
|
||||
'''A snazzy Doodad'''
|
||||
|
||||
__tablename__ = 'doodad'
|
||||
|
||||
doodad_id = Column(Integer(), primary_key=True)
|
||||
name = Column(String(255))
|
||||
purpose = Column(String(255))
|
||||
|
||||
def update(self, changes: DoodadInterface):
|
||||
for key, val in changes.items():
|
||||
setattr(self, key, val)
|
||||
return self
|
22
app/other_api/doodad/model_test.py
Normal file
22
app/other_api/doodad/model_test.py
Normal file
@@ -0,0 +1,22 @@
|
||||
from pytest import fixture
|
||||
from flask_sqlalchemy import SQLAlchemy
|
||||
from app.test.fixtures import app, db # noqa
|
||||
from .model import Doodad
|
||||
|
||||
|
||||
@fixture
|
||||
def doodad() -> Doodad:
|
||||
return Doodad(
|
||||
doodad_id=1, name='Test doodad', purpose='Test purpose'
|
||||
)
|
||||
|
||||
|
||||
def test_Doodad_create(doodad: Doodad):
|
||||
assert doodad
|
||||
|
||||
|
||||
def test_Doodad_retrieve(doodad: Doodad, db: SQLAlchemy): # noqa
|
||||
db.session.add(doodad)
|
||||
db.session.commit()
|
||||
s = Doodad.query.first()
|
||||
assert s.__dict__ == doodad.__dict__
|
9
app/other_api/doodad/schema.py
Normal file
9
app/other_api/doodad/schema.py
Normal file
@@ -0,0 +1,9 @@
|
||||
from marshmallow import fields, Schema
|
||||
|
||||
|
||||
class DoodadSchema(Schema):
|
||||
'''Doodad schema'''
|
||||
|
||||
doodadId = fields.Number(attribute='doodad_id')
|
||||
name = fields.String(attribute='name')
|
||||
purpose = fields.String(attribute='purpose')
|
0
app/other_api/doodad/schema_test.py
Normal file
0
app/other_api/doodad/schema_test.py
Normal file
41
app/other_api/doodad/service.py
Normal file
41
app/other_api/doodad/service.py
Normal file
@@ -0,0 +1,41 @@
|
||||
from app import db
|
||||
from typing import List
|
||||
from .model import Doodad
|
||||
from .interface import DoodadInterface
|
||||
|
||||
|
||||
class DoodadService():
|
||||
@staticmethod
|
||||
def get_all() -> List[Doodad]:
|
||||
return Doodad.query.all()
|
||||
|
||||
@staticmethod
|
||||
def get_by_id(doodad_id: int) -> Doodad:
|
||||
return Doodad.query.get(doodad_id)
|
||||
|
||||
@staticmethod
|
||||
def update(doodad: Doodad, Doodad_change_updates: DoodadInterface) -> Doodad:
|
||||
doodad.update(Doodad_change_updates)
|
||||
db.session.commit()
|
||||
return doodad
|
||||
|
||||
@staticmethod
|
||||
def delete_by_id(doodad_id: int) -> List[int]:
|
||||
doodad = Doodad.query.filter(Doodad.doodad_id == doodad_id).first()
|
||||
if not doodad:
|
||||
return []
|
||||
db.session.delete(doodad)
|
||||
db.session.commit()
|
||||
return [doodad_id]
|
||||
|
||||
@staticmethod
|
||||
def create(new_attrs: DoodadInterface) -> Doodad:
|
||||
new_doodad = Doodad(
|
||||
name=new_attrs['name'],
|
||||
purpose=new_attrs['purpose']
|
||||
)
|
||||
|
||||
db.session.add(new_doodad)
|
||||
db.session.commit()
|
||||
|
||||
return new_doodad
|
60
app/other_api/doodad/service_test.py
Normal file
60
app/other_api/doodad/service_test.py
Normal file
@@ -0,0 +1,60 @@
|
||||
from flask_sqlalchemy import SQLAlchemy
|
||||
from typing import List
|
||||
from app.test.fixtures import app, db # noqa
|
||||
from .model import Doodad
|
||||
from .service import DoodadService # noqa
|
||||
from .interface import DoodadInterface
|
||||
|
||||
|
||||
def test_get_all(db: SQLAlchemy): # noqa
|
||||
yin: Doodad = Doodad(doodad_id=1, name='Yin', purpose='thing 1')
|
||||
yang: Doodad = Doodad(doodad_id=2, name='Yang', purpose='thing 2')
|
||||
db.session.add(yin)
|
||||
db.session.add(yang)
|
||||
db.session.commit()
|
||||
|
||||
results: List[Doodad] = DoodadService.get_all()
|
||||
|
||||
assert len(results) == 2
|
||||
assert yin in results and yang in results
|
||||
|
||||
|
||||
def test_update(db: SQLAlchemy): # noqa
|
||||
yin: Doodad = Doodad(doodad_id=1, name='Yin', purpose='thing 1')
|
||||
|
||||
db.session.add(yin)
|
||||
db.session.commit()
|
||||
updates: DoodadInterface = dict(name='New Doodad name')
|
||||
|
||||
DoodadService.update(yin, updates)
|
||||
|
||||
result: Doodad = Doodad.query.get(yin.doodad_id)
|
||||
assert result.name == 'New Doodad name'
|
||||
|
||||
|
||||
def test_delete_by_id(db: SQLAlchemy): # noqa
|
||||
yin: Doodad = Doodad(doodad_id=1, name='Yin', purpose='thing 1')
|
||||
yang: Doodad = Doodad(doodad_id=2, name='Yang', purpose='thing 2')
|
||||
db.session.add(yin)
|
||||
db.session.add(yang)
|
||||
db.session.commit()
|
||||
|
||||
DoodadService.delete_by_id(1)
|
||||
db.session.commit()
|
||||
|
||||
results: List[Doodad] = Doodad.query.all()
|
||||
|
||||
assert len(results) == 1
|
||||
assert yin not in results and yang in results
|
||||
|
||||
|
||||
def test_create(db: SQLAlchemy): # noqa
|
||||
|
||||
yin: DoodadInterface = dict(name='Fancy new doodad', purpose='Fancy new purpose')
|
||||
DoodadService.create(yin)
|
||||
results: List[Doodad] = Doodad.query.all()
|
||||
|
||||
assert len(results) == 1
|
||||
|
||||
for k in yin.keys():
|
||||
assert getattr(results[0], k) == yin[k]
|
Reference in New Issue
Block a user