<s> from datetime import datetime , timedelta <EOL> from flask import g , render_template , request , jsonify , make_response <EOL> from flask . ext . sqlalchemy import SQLAlchemy <EOL> from sqlalchemy . orm import relationship <EOL> from flask_oauthlib . provider import OAuth2Provider <EOL> from flask_oauthlib . contrib . oauth2 import bind_sqlalchemy <EOL> from flask_oauthlib . contrib . oauth2 import bind_cache_grant <EOL> db = SQLAlchemy ( ) <EOL> class User ( db . Model ) : <EOL> id = db . Column ( db . Integer , primary_key = True ) <EOL> username = db . Column ( db . String ( <NUM_LIT> ) , unique = True , index = True , <EOL> nullable = False ) <EOL> def check_password ( self , password ) : <EOL> return True <EOL> class Client ( db . Model ) : <EOL> name = db . Column ( db . String ( <NUM_LIT> ) ) <EOL> client_id = db . Column ( db . String ( <NUM_LIT> ) , primary_key = True ) <EOL> client_secret = db . Column ( db . String ( <NUM_LIT> ) , unique = True , index = True , <EOL> nullable = False ) <EOL> client_type = db . Column ( db . String ( <NUM_LIT:20> ) , default = '<STR_LIT>' ) <EOL> _redirect_uris = db . Column ( db . Text ) <EOL> default_scope = db . Column ( db . Text , default = '<STR_LIT>' ) <EOL> @ property <EOL> def user ( self ) : <EOL> return User . query . get ( <NUM_LIT:1> ) <EOL> @ property <EOL> def redirect_uris ( self ) : <EOL> if self . _redirect_uris : <EOL> return self . _redirect_uris . split ( ) <EOL> return [ ] <EOL> @ property <EOL> def default_redirect_uri ( self ) : <EOL> return self . redirect_uris [ <NUM_LIT:0> ] <EOL> @ property <EOL> def default_scopes ( self ) : <EOL> if self . default_scope : <EOL> return self . default_scope . split ( ) <EOL> return [ ] <EOL> @ property <EOL> def allowed_grant_types ( self ) : <EOL> return [ '<STR_LIT>' , '<STR_LIT:password>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> class Grant ( db . Model ) : <EOL> id = db . Column ( db . Integer , primary_key = True ) <EOL> user_id = db . Column ( <EOL> db . Integer , db . ForeignKey ( '<STR_LIT>' , ondelete = '<STR_LIT>' ) <EOL> ) <EOL> user = relationship ( '<STR_LIT>' ) <EOL> client_id = db . Column ( <EOL> db . String ( <NUM_LIT> ) , db . ForeignKey ( '<STR_LIT>' , ondelete = '<STR_LIT>' ) , <EOL> nullable = False , <EOL> ) <EOL> client = relationship ( '<STR_LIT>' ) <EOL> code = db . Column ( db . String ( <NUM_LIT:255> ) , index = True , nullable = False ) <EOL> redirect_uri = db . Column ( db . String ( <NUM_LIT:255> ) ) <EOL> scope = db . Column ( db . Text ) <EOL> expires = db . Column ( db . DateTime ) <EOL> def delete ( self ) : <EOL> db . session . delete ( self ) <EOL> db . session . commit ( ) <EOL> return self <EOL> @ property <EOL> def scopes ( self ) : <EOL> if self . scope : <EOL> return self . scope . split ( ) <EOL> return None <EOL> class Token ( db . Model ) : <EOL> id = db . Column ( db . Integer , primary_key = True ) <EOL> client_id = db . Column ( <EOL> db . String ( <NUM_LIT> ) , db . ForeignKey ( '<STR_LIT>' , ondelete = '<STR_LIT>' ) , <EOL> nullable = False , <EOL> ) <EOL> user_id = db . Column ( <EOL> db . Integer , db . ForeignKey ( '<STR_LIT>' , ondelete = '<STR_LIT>' ) <EOL> ) <EOL> user = relationship ( '<STR_LIT>' ) <EOL> client = relationship ( '<STR_LIT>' ) <EOL> token_type = db . Column ( db . String ( <NUM_LIT> ) ) <EOL> access_token = db . Column ( db . String ( <NUM_LIT:255> ) ) <EOL> refresh_token = db . Column ( db . String ( <NUM_LIT:255> ) ) <EOL> expires = db . Column ( db . DateTime ) <EOL> scope = db . Column ( db . Text ) <EOL> def __init__ ( self , ** kwargs ) : <EOL> expires_in = kwargs . pop ( '<STR_LIT>' ) <EOL> self . expires = datetime . utcnow ( ) + timedelta ( seconds = expires_in ) <EOL> for k , v in kwargs . items ( ) : <EOL> setattr ( self , k , v ) <EOL> @ property <EOL> def scopes ( self ) : <EOL> if self . scope : <EOL> return self . scope . split ( ) <EOL> return [ ] <EOL> def delete ( self ) : <EOL> db . session . delete ( self ) <EOL> db . session . commit ( ) <EOL> return self <EOL> def current_user ( ) : <EOL> return g . user <EOL> def cache_provider ( app ) : <EOL> oauth = OAuth2Provider ( app ) <EOL> bind_sqlalchemy ( oauth , db . session , user = User , <EOL> token = Token , client = Client ) <EOL> app . config . update ( { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> bind_cache_grant ( app , oauth , current_user ) <EOL> return oauth <EOL> def sqlalchemy_provider ( app ) : <EOL> oauth = OAuth2Provider ( app ) <EOL> bind_sqlalchemy ( oauth , db . session , user = User , token = Token , <EOL> client = Client , grant = Grant , current_user = current_user ) <EOL> return oauth <EOL> def default_provider ( app ) : <EOL> oauth = OAuth2Provider ( app ) <EOL> @ oauth . clientgetter <EOL> def get_client ( client_id ) : <EOL> return Client . query . filter_by ( client_id = client_id ) . first ( ) <EOL> @ oauth . grantgetter <EOL> def get_grant ( client_id , code ) : <EOL> return Grant . query . filter_by ( client_id = client_id , code = code ) . first ( ) <EOL> @ oauth . tokengetter <EOL> def get_token ( access_token = None , refresh_token = None ) : <EOL> if access_token : <EOL> return Token . query . filter_by ( access_token = access_token ) . first ( ) <EOL> if refresh_token : <EOL> return Token . query . filter_by ( refresh_token = refresh_token ) . first ( ) <EOL> return None <EOL> @ oauth . grantsetter <EOL> def set_grant ( client_id , code , request , * args , ** kwargs ) : <EOL> expires = datetime . utcnow ( ) + timedelta ( seconds = <NUM_LIT:100> ) <EOL> grant = Grant ( <EOL> client_id = client_id , <EOL> code = code [ '<STR_LIT:code>' ] , <EOL> redirect_uri = request . redirect_uri , <EOL> scope = '<STR_LIT:U+0020>' . join ( request . scopes ) , <EOL> user_id = g . user . id , <EOL> expires = expires , <EOL> ) <EOL> db . session . add ( grant ) <EOL> db . session . commit ( ) <EOL> @ oauth . tokensetter <EOL> def set_token ( token , request , * args , ** kwargs ) : <EOL> tok = Token ( ** token ) <EOL> tok . user_id = request . user . id <EOL> tok . client_id = request . client . client_id <EOL> db . session . add ( tok ) <EOL> db . session . commit ( ) <EOL> @ oauth . usergetter <EOL> def get_user ( username , password , * args , ** kwargs ) : <EOL> return User . query . filter_by ( username = username ) . first ( ) <EOL> return oauth <EOL> def prepare_app ( app ) : <EOL> db . init_app ( app ) <EOL> db . app = app <EOL> db . create_all ( ) <EOL> client1 = Client ( <EOL> name = '<STR_LIT>' , client_id = '<STR_LIT>' , client_secret = '<STR_LIT>' , <EOL> _redirect_uris = ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> ) , <EOL> ) <EOL> client2 = Client ( <EOL> name = '<STR_LIT>' , client_id = '<STR_LIT>' , <EOL> client_secret = '<STR_LIT>' , client_type = '<STR_LIT>' , <EOL> _redirect_uris = ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> ) , <EOL> ) <EOL> user = User ( username = '<STR_LIT>' ) <EOL> temp_grant = Grant ( <EOL> user_id = <NUM_LIT:1> , client_id = '<STR_LIT>' , <EOL> code = '<STR_LIT>' , scope = '<STR_LIT:email>' , <EOL> expires = datetime . utcnow ( ) + timedelta ( seconds = <NUM_LIT:100> ) <EOL> ) <EOL> access_token = Token ( <EOL> user_id = <NUM_LIT:1> , client_id = '<STR_LIT>' , access_token = '<STR_LIT>' , expires_in = <NUM_LIT:0> <EOL> ) <EOL> try : <EOL> db . session . add ( client1 ) <EOL> db . session . add ( client2 ) <EOL> db . session . add ( user ) <EOL> db . session . add ( temp_grant ) <EOL> db . session . add ( access_token ) <EOL> db . session . commit ( ) <EOL> except : <EOL> db . session . rollback ( ) <EOL> return app <EOL> def create_server ( app , oauth = None ) : <EOL> if not oauth : <EOL> oauth = default_provider ( app ) <EOL> app = prepare_app ( app ) <EOL> @ app . before_request <EOL> def load_current_user ( ) : <EOL> user = User . query . get ( <NUM_LIT:1> ) <EOL> g . user = user <EOL> @ app . route ( '<STR_LIT>' ) <EOL> def home ( ) : <EOL> return render_template ( '<STR_LIT>' ) <EOL> @ app . route ( '<STR_LIT>' , methods = [ '<STR_LIT:GET>' , '<STR_LIT:POST>' ] ) <EOL> @ oauth . authorize_handler <EOL> def authorize ( * args , ** kwargs ) : <EOL> if request . method == '<STR_LIT:GET>' : <EOL> return render_template ( '<STR_LIT>' ) <EOL> if request . method == '<STR_LIT>' : <EOL> response = make_response ( '<STR_LIT>' , <NUM_LIT:200> ) <EOL> response . headers [ '<STR_LIT>' ] = kwargs . get ( '<STR_LIT>' ) <EOL> return response <EOL> confirm = request . form . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> return confirm == '<STR_LIT:yes>' <EOL> @ app . route ( '<STR_LIT>' , methods = [ '<STR_LIT:POST>' , '<STR_LIT:GET>' ] ) <EOL> @ oauth . token_handler <EOL> def access_token ( ) : <EOL> return { } <EOL> @ app . route ( '<STR_LIT>' , methods = [ '<STR_LIT:POST>' ] ) <EOL> @ oauth . revoke_handler <EOL> def revoke_token ( ) : <EOL> pass <EOL> @ app . route ( '<STR_LIT>' ) <EOL> @ oauth . require_oauth ( '<STR_LIT:email>' ) <EOL> def email_api ( ) : <EOL> oauth = request . oauth <EOL> return jsonify ( email = '<STR_LIT>' , username = oauth . user . username ) <EOL> @ app . route ( '<STR_LIT>' ) <EOL> @ oauth . require_oauth ( ) <EOL> def client_api ( ) : <EOL> oauth = request . oauth <EOL> return jsonify ( client = oauth . client . name ) <EOL> @ app . route ( '<STR_LIT>' ) <EOL> @ oauth . require_oauth ( '<STR_LIT:address>' ) <EOL> def address_api ( city ) : <EOL> oauth = request . oauth <EOL> return jsonify ( address = city , username = oauth . user . username ) <EOL> @ app . route ( '<STR_LIT>' , methods = [ '<STR_LIT:GET>' , '<STR_LIT:POST>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> @ oauth . require_oauth ( ) <EOL> def method_api ( ) : <EOL> return jsonify ( method = request . method ) <EOL> @ oauth . invalid_response <EOL> def require_oauth_invalid ( req ) : <EOL> return jsonify ( message = req . error_message ) , <NUM_LIT> <EOL> return app <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> from flask import Flask <EOL> app = Flask ( __name__ ) <EOL> app . debug = True <EOL> app . secret_key = '<STR_LIT>' <EOL> app . config . update ( { <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } ) <EOL> app = create_server ( app ) <EOL> app . run ( ) </s>
