User Management API Implementation

This article demonstrates how to implement CRUD operations for a user management module using Flask. The examples focus on comon patterns encountered in production systems.

Environment: Python 3.6, PostgreSQL 10, Flask 0.12

Create User Endpoint

Route Definition

@app.route('/guardian/users', methods=['POST'])
@require_role('guardian_admin')
@login_required
def create_user():
    """
    Expected JSON payload:
    {
        "user": {
            "domain_id": "default",
            "enabled": True,
            "name": "testuser",
            "password": "secret123",
            "realname": "John Doe",
            "email": "john@example.com",
            "tel": "+86-029-88880000",
            "phone": "13800138000",
            "org_id": "7d22d25a-b648-4ad3-a834-2398eb989a9e",
            "org_array": [],
            "comments": "User notes"
        }
    }
    """
    try:
        domain_id = None
        user_name = None
        user_passwd = None
        project_id = None
        real_name = None
        email = None
        phone = None
        org_id = None
        org_array = None
        comments = None
        
        req_obj = request.data
        req_json = req_obj.decode() if isinstance(req_obj, bytes) else req_obj
        user_obj = jsonpickle.loads(req_json)
        
        if 'default_project_id' in user_obj['user']:
            project_id = user_obj['user']['default_project_id']
        
        domain_id = user_obj['user']['domain_id']
        user_name = user_obj['user']['name']
        user_passwd = user_obj['user']['password']
        
    except Exception as ee:
        abort(400)
    
    if domain_id and user_name and user_passwd:
        resp = usermgt.create_user(domain_id, user_name, user_passwd, project_id)
        if 'id' in resp['user']:
            try:
                user_id = resp['user']['id']
                real_name = user_obj['user'].get('realname')
                email = user_obj['user'].get('email')
                tel = user_obj['user'].get('tel')
                phone = user_obj['user'].get('phone')
                org_id = user_obj['user'].get('org_id')
                org_array = user_obj['user'].get('org_array')
                comments = user_obj['user'].get('comments')
                
                GuardianModelDao._save_update_users_info(
                    user_id, user_name, real_name, email, tel, phone, 
                    org_id, org_array, comments
                )
                return wrap_response(SUCCESS_MSG, 200)
            except Exception as e:
                abort(500)
        else:
            abort(500)

Database Layer

@classmethod
def _save_update_users_info(cls, user_id, username, real_name, email, tel, phone, org_id, org_array, comments):
    LOG.debug('start create users info...')
    try:
        with TxConn() as conn:
            with conn.cursor() as cursor:
                now_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
                users_data = {
                    "user_id": user_id,
                    "username": username,
                    "realname": real_name,
                    "email": email,
                    "tel": tel,
                    "phone": phone,
                    "org": org_id,
                    "org_array": org_array,
                    "comments": comments,
                    "now_time": now_time
                }
                cursor.execute(cls.insert_users_info, users_data)
            conn.commit()
    except Exception as e:
        LOG.exception('error saving users info', e)
        raise e

Test Scrript for Create User

import unittest
import requests

class GuardianTestCase(unittest.TestCase):
    guardian_base = 'http://127.0.0.1:7070'
    user_token = keystone.user_token('default', 'admin', 'password')
    
    def test_create_user_api(self):
        myheaders = {
            'Content-Type': 'application/json',
            'X-Subject-Token': self.user_token
        }
        
        user_payload = {
            "user": {
                "domain_id": "default",
                "enabled": True,
                "name": "testuser1",
                "password": "secret123",
                "realname": "John Doe",
                "email": "john@example.com",
                "tel": "+86-029-88880000",
                "phone": "13800138000",
                "org_id": "7d22d25a-b648-4ad3-a834-2398eb989a9e",
                "org_array": None,
                "comments": "Test user"
            }
        }
        
        response = requests.post(
            self.guardian_base + '/guardian/users',
            json=user_payload,
            headers=myheaders,
            verify=True
        )
        
        if response.ok:
            self.assertIsNotNone(response.content)
            LOG.info('create users, response: {0}'.format(response.content))
        else:
            LOG.error('error create users, response: {0}, status: {1}'.format(
                response.content, response.status_code))

Update User Endpoint

The update endpoint uses PATCH method with user_id as a path parameter:

@app.route('/guardian/users/<string:user_id>', methods=['PATCH'])
@require_role('guardian_admin')
@login_required
def update_user(user_id):
    try:
        resp = usermgt.update_user(domain_id, user_id, username, user_passwd, project_id, enabled)
        if resp is not None:
            GuardianModelDao._save_update_users_info(
                user_id, username, real_name, email, tel, phone, 
                org_id, org_array, comments
            )
            return wrap_response(SUCCESS_MSG, 200)
    except Exception as e:
        LOG.exception('error modify user', e)
        abort(500)

Test Script for Udpate User

def test_update_user_api(self):
    myheaders = {
        'Content-Type': 'application/json',
        'X-Subject-Token': self.user_token
    }
    
    user_payload = {
        "user": {
            "domain_id": "default",
            "enabled": True,
            "name": "testuser1",
            "password": "newpass123",
            "realname": "Jane Doe",
            "email": "jane@example.com",
            "tel": "+86-029-88880000",
            "phone": "13800138000",
            "org_id": "7d22d25a-b648-4ad3-a834-2398eb989a9e",
            "comments": "Updated user"
        }
    }
    
    response = requests.patch(
        self.guardian_base + '/guardian/users/0d26176579e94d558c4c2d864b003241',
        json=user_payload,
        headers=myheaders,
        verify=True
    )
    
    if response.ok:
        self.assertIsNotNone(response.content)
        LOG.info('update users, response: {0}'.format(response.content))
    else:
        LOG.error('error update users, response: {0}, status: {1}'.format(
            response.content, response.status_code))

Query User Endpoint

@app.route('/guardian/users/<string:user_name>', methods=['GET'])
@require_role('guardian_admin')
@login_required
def query_user_by_username(user_name):
    try:
        user_info = GuardianModelDao._query_users_info_by_name(user_name)
        if len(user_info):
            return wrap_response(jsonpickle.dumps(user_info), 200)
        else:
            return wrap_response("[]", 200)
    except Exception as e:
        LOG.exception('error query user by name', e)
        abort(500)

Database Query Layer

@classmethod
def _query_users_info_by_id(cls, user_id):
    LOG.debug('start list users info...')
    users = []
    with TxConn() as conn:
        try:
            with conn.cursor() as cursor:
                data = (user_id,)
                cursor.execute(cls.query_users_info_by_id, data)
                for (user_id, real_name, email, tel, phone, org_id, org_array, 
                     comments, create_time, update_time) in cursor:
                    user = Users(user_id, real_name, email, tel, phone, org_id, 
                                org_array, comments, create_time, update_time)
                    users.append(user)
        except Exception as e:
            LOG.exception('error query users !')
            raise e
    return users

Test Script for Query User

def test_query_user_api(self):
    myheaders = {
        'Content-Type': 'application/json',
        'X-Subject-Token': self.user_token
    }
    
    response = requests.get(
        self.guardian_base + '/guardian/users/testuser',
        headers=myheaders,
        verify=True
    )
    
    if response.ok:
        self.assertIsNotNone(response.text)
        LOG.info('query users, response: {0}'.format(response.text))
    else:
        LOG.error('error query users, response: {0}, status: {1}'.format(
            response.text, response.status_code))

Delete User Endpoint

@app.route('/guardian/users/<string:user_id>', methods=['DELETE'])
@require_role('guardian_admin')
@login_required
def delete_user(user_id):
    if user_id is None:
        abort(400)
    try:
        status_code = usermgt.delete_user(user_id)
        if status_code == 204:
            return wrap_response(SUCCESS_MSG, status=status_code)
    except Exception as e:
        LOG.exception('error delete users info', e)
        abort(500)

Backend Delete Logic

def delete_user(user_id):
    myheaders = {
        'Content-Type': 'application/json',
        'X-Auth-Token': settings.KEYSTONE_ADMIN_TOKEN
    }
    response = requests.delete(
        settings.KEYSTONE_BASE + '/v3/users/' + user_id,
        headers=myheaders,
        verify=False
    )
    
    if response.status_code == 204:
        LOG.debug('delete users -- response {0}'.format(response))
        return response.status_code
    else:
        LOG.error('error list users, response: {0}'.format(response))
        raise KeystoneException(
            'error delete user {0} with response status {1}'.format(
                user_id, response.status_code), 
            response.status_code
        )

Test Script for Delete User

def test_delete_users_api(self):
    myheaders = {
        'Content-Type': 'application/json',
        'X-Subject-Token': self.user_token
    }
    
    response = requests.delete(
        self.guardian_base + '/guardian/users/testuser1',
        headers=myheaders,
        verify=True
    )
    
    if response.ok:
        self.assertIsNotNone(response.content)
        LOG.info('delete users api, response: {0}'.format(response.content))
    else:
        LOG.error('error delete users api, response: {0}, status: {1}'.format(
            response.content, response.status_code))

API Documentation Summary

Create User Request Body:

{
    "user": {
        "domain_id": "default",
        "enabled": true,
        "name": "testuser",
        "password": "secret123",
        "realname": "John Doe",
        "email": "john@example.com",
        "tel": "+86-029-88880000",
        "phone": "13800138000",
        "org_id": "7d22d25a-b648-4ad3-a834-2398eb989a9e",
        "org_array": null,
        "comments": "User notes"
    }
}

Update User Request Body:

{
    "user": {
        "domain_id": "default",
        "enabled": true,
        "name": "testuser",
        "password": "newsecret",
        "realname": "Jane Doe",
        "email": "jane@example.com",
        "tel": "+86-029-88880000",
        "phone": "13800138000",
        "org_id": "7d22d25a-b648-4ad3-a834-2398eb989a9e",
        "org_array": null,
        "comments": "Updated notes"
    }
}