Completion of the web service

app/db_models.py

@@ -3,20 +3,7 @@
 from peewee import *
 
 
-config = configparser.ConfigParser()
-config.read('../config.ini')
-
-params_for_db = {'host': config['db']['host'],
-                 'user': config['db']['user'],
-                 'password': config['db']['password'],
-                 'database': config['db']['database']}
-
-#if is test
-test_database = config['db']['test_database']
-if test_database:
-    params_for_db['database'] = test_database
-
-db = PostgresqlDatabase(**params_for_db)
+db = PostgresqlDatabase(database='postgres', user='postgres')
 db.connect()
 
 
@@ -36,6 +23,6 @@ class Employee(BaseModel):
 
 
 
-if __name__ == '__main__' or config:
+if __name__ == '__main__':
     db.drop_tables([Department, Employee])
     db.create_tables([Department, Employee])

app/main.py

@@ -1,8 +1,10 @@
-from flask import Flask, request
-from flask_restful import Api, Resource
+import peewee
 
-from app.db_models import Employee, Department
+from datetime import datetime
+from flask import Flask, request, Response, jsonify
+from flask_restful import Api, Resource
 
+from db_models import Employee, Department
 
 app = Flask(__name__)
 api = Api(app)
@@ -14,40 +16,151 @@ def new_func(*args, **kwargs):
             return f(*args, **kwargs)
         else:
             return {}, 415
+
     return new_func
-
-
-class DepartmentApi(Resource):
+
+
+def obj_to_dict(obj):
+    return {field: getattr(obj, field) for field in obj._meta.fields}
+
+
+class DepartmentsApi(Resource):
     def get(self):
-        departments = Department.select(Department.name_department)
-        return {'data': [dep.name_department
-                            for dep in departments]}, 200
+        departments = Department.select()
+
+        return {'data': [obj_to_dict(dep)
+                         for dep in departments]
+                }, 200
+
     @is_json
     def post(self):
         try:
             department = Department.create(**request.json)
-        except IntegrityError:
-            return {}, 409
+
+        except peewee.DataError:
+            return {'error': 'Invalid data type'}, 415
+
+        except peewee.IntegrityError:
+            return {'error': "It's a duplicate"}, 409
+
         else:
+            dep_dict = request.json
+            dep_dict['id'] = department.id
             department.save()
-            return {'data': request.json}
-
-    @is_json
-    def delete(self):
-        name = request.json.get('name')
-        if name is None:
-            return {}, 400
-        department = Department.delete().where(
-            Department.name_department == name
-        )
-        department.execute()
+            return {'data': dep_dict}
+
+
+class DepartmentApi(Resource):
+    def get(self, dep_id):
+        dep = Department.select().where(Department.id == dep_id)
+        try:
+            dep = dep[0]
+        except IndexError:
+            return {'error': 'Department not found'}, 404
+        else:
+
+            return {'data': {'id': dep.id,
+                             'name_department': dep.name_department
+                             }
+                    }, 200
+
+    def delete(self, dep_id):
+        Department.delete_by_id(dep_id)
         return {}, 200
 
 
-
-api.add_resource(DepartmentApi, '/department')
+def emp_to_dict(emp):
+    emp_dict = {'id': emp.id,
+                'name_department': emp.name_department.name_department,
+                'full_name': emp.full_name,
+                'date_of_brith': emp.date_of_brith.isoformat(),
+                'salary': emp.salary
+                }
+    return emp_dict
+
+
+class EmployeesApi(Resource):
+    def get(self):
+        employees = Employee.select().join(Department)
+
+        data = {'data': [emp_to_dict(emp) for emp in employees]}
+
+        if not len(data['data']):
+            return {'error': 'No employees'}, 404
+
+        return data, 200
+
+    @is_json
+    def post(self):
+
+        emp_dict = {field: request.json.get(field) for field in ['full_name',
+                                                                 'name_department',
+                                                                 'date_of_brith',
+                                                                 'salary']}
+
+        if None in emp_dict.values():
+            return {'error': 'One of the parameters is missing or it is not'}, 400
+
+        try:
+            employee = Employee.create(**emp_dict)
+
+        except peewee.DataError:
+            return {'error': 'Invalid data type'}, 415
+
+        except peewee.IntegrityError:
+            return {}, 409
+        except:
+            return {}, 500
+        else:
+            emp_dict['id'] = employee.id
+            employee.save()
+            return {'data': emp_dict}, 200
+
+
+class EmployeeApi(Resource):
+    def get(self, emp_id):
+        emp = Employee.select().where(Employee.id == emp_id)
+        try:
+            emp = emp[0]
+        except IndexError:
+            'Employee not found'
+        else:
+            return {'data': emp_to_dict(emp)}, 200
+
+    @is_json
+    def put(self, emp_id):
+        emp_dict = {field: request.json.get(field) for field in ['full_name',
+                                                                 'name_department',
+                                                                 'date_of_brith',
+                                                                 'salary']}
+        if emp_dict['date_of_brith'] is not None:
+            return {'error': 'Date of birth cannot be changed'}, 400
+
+        emp_dict_keys = list(emp_dict.keys())
+        for k in emp_dict_keys:
+            if emp_dict[k] is None:
+                emp_dict.pop(k)
+
+        count_of_updates = Employee.update(emp_dict).where(
+            Employee.id == emp_id).execute()
+
+        if count_of_updates == 0:
+            return {'error': 'Employee not found'}, 404
+
+        emp = Employee.select().where(Employee.id == emp_id)[0]
+
+        return {'data': emp_to_dict(emp)}, 200
+
+    def delete(self, emp_id):
+        Employee.delete_by_id(emp_id)
+        return {}, 200
+
 
+api.add_resource(DepartmentsApi, '/department/')
+api.add_resource(DepartmentApi, '/department/<int:dep_id>')
+api.add_resource(EmployeesApi, '/employee/')
+api.add_resource(EmployeeApi, '/employee/<int:emp_id>')
 
 if __name__ == '__main__':
-    connect(database='postgres', user='postgres')
-    app.run(debug=True)
+    # connect(database='postgres', user='postgres')
+    app.run()

app/test.py

@@ -1,46 +1,37 @@
-import unittest
-import psycopg2
-import configparser
-
-
-class MyTestCase(unittest.TestCase):
-    def setUp(self):
-        self.config = config = configparser.ConfigParser()
-        config.read('../config.ini')
-
-        # write test config
-        config['db']['test_database'] = 'test_epam_app'
-
-        with open('../config.ini', 'w') as f:
-            config.write(f)
-
-        params_for_db = {'host': config['db']['host'],
-                         'user': config['db']['user'],
-                         'password': config['db']['password'],
-                         'database': config['db']['database']}
-
-
-        # create test DB
-        with psycopg2.connect(**params_for_db) as connect:
-            connect.set_isolation_level(0)
-            with connect.cursor() as cursor:
-                try:
-                    cursor.execute('drop database test_epam_app')
-                finally:
-                    cursor.execute('create database test_epam_app')
-
-
-        from app.main import app
-        app.testing = True
-        self.app = app.test_client()
-
-    def tearDown(self):
-        self.config['db']['test_database'] = ''
-        with open('../config.ini', 'w') as f:
-            self.config.write(f)
-
-    def test_home(self):
-        result = self.app.get('/department')
-        print(type(result))
-
-unittest.main()
+import requests
+
+url = 'http://127.0.0.1:5000/'
+
+
+# DEPARTMENT TEST
+def dep_test():
+    get_dep = requests.get(url + 'department').json()
+    print(get_dep)
+
+    id = requests.post(url + 'department',
+                             json={'name_department': 'Прогроммист'}).json()['data']['id']
+    print(id)
+    del_dep = requests.delete(url + 'department/' + str(id)).json()
+    print(del_dep)
+
+##################
+# EMPLOYEE TEST
+def emp_test():
+    get_emp = requests.get(url + 'employee').json()
+    #print(get_emp)
+
+    post_emp = requests.post(url + 'employee',
+                             json={'name_department': 'Повар',
+                                   'full_name': 'Edgar Gorobchuk',
+                                   'date_of_brith': '2001-05-28',
+                                   'salary': 100500}).json()
+    print(post_emp)
+    print('=====')
+
+    put_emp = requests.put(url + 'employee/' + '40',
+                           json={'full_name': 'Me Me'}).json()
+    print(put_emp)
+
+    requests.delete(url + 'employee/' + '2')
+
+emp_test()