forked from thomaxxl/safrs
-
Notifications
You must be signed in to change notification settings - Fork 0
/
demo_hashid.py
91 lines (74 loc) · 2.68 KB
/
demo_hashid.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
#!/usr/bin/env python
#
# This is a demo application to demonstrate the functionality of the safrs_rest REST API
#
# It can be ran standalone like this:
# python demo.py [Listener-IP]
#
# This will run the example on http://Listener-Ip:5000
#
# - A database is created and a user is added
# - A rest api is available
# - swagger2 documentation is generated
#
import sys
if sys.version_info[0] == 3:
import builtins as __builtin__
else:
import __builtin__
from flask import Flask, redirect
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import Column, Integer, String
from safrs.db import SAFRSBase, documented_api_method, SAFRSSHA256HashID
from safrs.jsonapi import SAFRSRestAPI, SAFRSJSONEncoder, Api
from flask_swagger_ui import get_swaggerui_blueprint
from flask_marshmallow import Marshmallow
app = Flask('demo_app')
app.config.update( SQLALCHEMY_DATABASE_URI = 'sqlite://',
DEBUG = True)
db = SQLAlchemy(app)
# Example sqla database object
class User(SAFRSBase, db.Model):
'''
description: User description
'''
__tablename__ = 'users'
id = Column(String, primary_key=True)
name = Column(String, default = '')
email = Column(String, default = '')
id_type = SAFRSSHA256HashID
# Following method is exposed through the REST API
# This means it can be invoked with a HTTP POST
@documented_api_method
def send_mail(self, email):
'''
description : Send an email
args:
email:
type : string
example : test email
'''
content = 'Mail to {} : {}\n'.format(self.name, email)
with open('/tmp/mail.txt', 'a+') as mailfile :
mailfile.write(content)
return { 'result' : 'sent {}'.format(content)}
HOST = sys.argv[1] if len(sys.argv) > 1 else '0.0.0.0'
PORT = 5000
# Create the database
db.create_all()
with app.app_context():
# Create a user
user = User(name='test',email='em@il')
api = Api(app, api_spec_url = '/api/swagger', host = '{}:{}'.format(HOST,PORT), schemes = [ "http" ] )
# Expose the User object
api.expose_object(User)
# Set the JSON encoder used for object to json marshalling
app.json_encoder = SAFRSJSONEncoder
# Register the API at /api/docs
swaggerui_blueprint = get_swaggerui_blueprint('/api', '/api/swagger.json')
app.register_blueprint(swaggerui_blueprint, url_prefix='/api')
@app.route('/')
def goto_api():
return redirect('/api')
print('Starting API: http://{}:{}/api'.format(HOST,PORT))
#app.run(host=HOST, port = PORT)