Python flask_login 模块，LoginManager() 实例源码

def init_login_manager(db):
    """Init security extensions (login manager and principal)

    :param db: Database which stores user accounts and roles
    :type db: ``flask_sqlalchemy.SQLAlchemy``
    :return: Login manager and principal extensions
    :rtype: (``flask_login.LoginManager``, ``flask_principal.Principal``
    """
    login_manager = flask_login.LoginManager()
    principals = flask_principal.Principal()
    login_manager.anonymous_user = Anonymous

    @login_manager.unauthorized_handler
    def unauthorized():
        flask.abort(403)

    @login_manager.user_loader
    def load_user(user_id):
        return db.session.query(UserAccount).get(int(user_id))

    return login_manager, principals

def setUp(self):
        app = flask.Flask(__name__)
        app.config['TESTING'] = True
        app.config['KRB5_SERVICE_NAME'] = 'HTTP'
        app.config['KRB5_HOSTNAME'] = 'example.org'

        login_manager = flask_login.LoginManager(app)
        manager = flask_kerberos_login.KerberosLoginManager(app)

        @manager.save_user
        def save_user(peer_name):
            user = User(str(peer_name))
            # persist our user to the login manager
            login_manager.reload_user(user)

        @app.route('/')
        @flask_login.login_required
        def index():
            return flask_login.current_user.id

        self.app = app
        self.manager = manager

def test_flask_login_user_login(app):
    # LoginManager could be set up in app fixture in conftest.py instead
    login_manager = LoginManager()
    login_manager.init_app(app)

    # TODO: once event is marked, user id exists in MonthEvents and test will
    #       continue to pass, regardless of continued success; set to current
    #       microsecond to temporarily circumvent, but there should be a better
    #       way to fix user_id assignment (or tear down redis or something)
    user_id = datetime.now().microsecond

    with app.test_request_context():
        # set up and log in user
        user = User()
        user.id = user_id
        login_user(user)

        # test that user was logged in
        assert current_user.is_active
        assert current_user.is_authenticated
        assert current_user == user

        # test that user id was marked with 'user:logged_in' event
        assert user_id in MonthEvents('user:logged_in', now.year, now.month)

def _configure_login(self):
        login_manager = LoginManager()
        login_manager.init_app(app)

        @login_manager.user_loader
        def load_token(token):
            try:
                response = AuthClient().token.get(token)
            except HTTPError:
                return None
            except requests.ConnectionError:
                logger.warning('Wazo authentication server connection error')
                return None
            token = response.get('token')
            if not token:
                return None
            return UserUI(token, response.get('auth_id'))

def init_login(app):
    login_manager = LoginManager()
    login_manager.init_app(app)

    @login_manager.user_loader
    def load_user(user_id):
        if request.host.split('.', 1)[0] == 'admin':
            from .models import Admin
            return Admin.query.get(user_id)
        from .models import User
        return User.query.get(user_id)

    def unauthorized():
        return redirect(url_for('admin_security.login'))

    login_manager.unauthorized = unauthorized

def __init__(self, server):
        """Initialize the web server.

        Args:
            server: NetServer object      
        """
        self.server = server
        self.port = None

        # Create Flask app and configure
        self.app = Flask(__name__)
        self.app.config['ERROR_404_HELP'] = False

        # Create REST API instance
        self.restapi = RestApi(self.app, self.server)

        # Create LoginManager instance and configure
        self.login = LoginManager()
        self.login.init_app(self.app)
        self.login.request_loader(self.load_user)

def init_app(app):
    app.config.update({'SECRET_KEY': const.SECRET_KEY})

    login_manager = LoginManager(app)
    login_manager.session_protection = 'basic'
    login_manager.login_view = 'auth.xauth'
    login_manager.refresh_view = 'auth.xauth'

    class FFAuth(Object):
        pass

    @login_manager.user_loader
    def load_user(user_id):
        try:
            return FFAuth.query.get(user_id)
        except LeanCloudError as _:
            pass
        return None

    app.register_blueprint(main_view)
    app.register_blueprint(auth_view)

def setUp(self):
        self.app = Flask(__name__)
        self.app.config["SECRET_KEY"] = "test"
        self.app.config["WTF_CSRF_ENABLED"] = False
        self.app.config['TESTING'] = True
        self.app.config["FILEUPLOAD_ALLOWED_EXTENSIONS"] = ["png", "jpg"]
        self.app.config["FILEUPLOAD_CONVERT_TO_SNAKE_CASE"] = True

        self.login_manager = LoginManager(self.app)
        self.ffu = FlaskFileUpload(self.app)

        self.dummy_stream = io.BytesIO(b"some initial text data")
        self.fs = FileStorage(self.dummy_stream, "dummy.png")

def setUp(self):
        self.app = Flask(__name__)
        self.app.config["SECRET_KEY"] = "test"
        self.app.config["WTF_CSRF_ENABLED"] = False
        self.app.config['TESTING'] = True

        self.login_manager = LoginManager(self.app)
        self.ffu = FlaskFileUpload(self.app)

        self.client = self.app.test_client()

def init_app(app):
        """
        Wrapper for the flask_login method ``LoginManager::init_app``.

        Args:
            Flask app (Flask): The Flask app to initialize with the
                login manager
        """
        return login_manager.init_app(app)

def test_flask_login_user_logout(app):
    login_manager = LoginManager()
    login_manager.init_app(app)

    user_id = datetime.now().microsecond

    with app.test_request_context():
        # set up, log in, and log out user
        user = User()
        user.id = user_id
        login_user(user)
        logout_user()

        # test that user was logged out
        assert not current_user.is_active
        assert not current_user.is_authenticated
        assert not current_user == user

        # test that user id was marked with 'user:logged_out' event
        assert user_id in MonthEvents('user:logged_out', now.year, now.month)


# SQLALCHEMY

# TODO: Instead of sqlalchemy fixture (return: db, User),
#       each test could use sqlalchemy fixture (return:
#       db) and sqlalchemy_user fixture (return: User);
#       tests should use whichever is better practice.

def __init__(self):
        # self.google_host = get_config_param('host')
        self.login_manager = flask_login.LoginManager()
        self.login_manager.login_view = 'airflow.login'
        self.flask_app = None
        self.google_oauth = None
        self.api_rev = None

def __init__(self):
        self.ghe_host = get_config_param('host')
        self.login_manager = flask_login.LoginManager()
        self.login_manager.login_view = 'airflow.login'
        self.flask_app = None
        self.ghe_oauth = None
        self.api_rev = None

def init_login():
    login_manager = login.LoginManager()
    login_manager.init_app(app)

    @login_manager.user_loader
    def load_user(user_id):
        return db.session.query(User).get(user_id)

def register_login(app):
    login_manager = LoginManager()
    login_manager.init_app(app)
    login_manager.login_view = "auth.login"
    login_manager.session_protection = "strong"
    login_manager.login_message = _("Please login to access this page.")

    return login_manager

def register_login():
    login_manager = LoginManager()
    login_manager.login_view = "auth.login"
    login_manager.session_protection = "basic"
    login_manager.login_message = "Please login to access this page."

    @login_manager.user_loader
    def user_loader(id):
        from .server.models import User
        user = User.query.get(int(id))
        return user

    @login_manager.request_loader
    def user_loader_from_request(request):
        from .server.models import User
        config = current_app.config
        token = request.headers.get(config['LOGIN_TOKEN_HEADER'], None)
        if not token:
            token = request.args.get(config['LOGIN_TOKEN'], None)
        if token:
            user = User.check_api_key(token)
            if user:
                login_user(user=user, remember=True)
                return user

    @login_manager.unauthorized_handler
    def unauthorized():
        abort(403)

    return login_manager

def __init__(self):
        # self.google_host = get_config_param('host')
        self.login_manager = flask_login.LoginManager()
        self.login_manager.login_view = 'airflow.login'
        self.flask_app = None
        self.google_oauth = None
        self.api_rev = None

def __init__(self):
        self.ghe_host = get_config_param('host')
        self.login_manager = flask_login.LoginManager()
        self.login_manager.login_view = 'airflow.login'
        self.flask_app = None
        self.ghe_oauth = None
        self.api_rev = None

def init_extensions(app):
    from flask_login import LoginManager
    from flask_migrate import Migrate

    login = LoginManager(app=app)

    @login.user_loader
    def load_user(user_id):
        from graygram.models.user import User
        return User.query.get(user_id)

    Migrate(app, db)

def config_login_manager(app):
    """Configures the Flask-Login manager, used for the web endpoints."""

    login_manager = flask_login.LoginManager()
    login_manager.init_app(app)
    login_manager.login_view = "users.login"
    login_manager.login_message = ''
    login_manager.anonymous_user = AnonymousUser
    # noinspection PyTypeChecker
    login_manager.user_loader(_load_user)

    return login_manager

def init_login(app):
    login_manager = login.LoginManager()
    login_manager.init_app(app)

    # Create user loader function
    @login_manager.user_loader
    def load_user(user_id):
        return User.objects(id=user_id).first()

def create_app(config_name):
    """
    Main app settings
    """

    app = Flask(__name__, instance_relative_config=True)
    app.config.from_object(app_config[config_name])
    app.config.from_pyfile('config.py')
    db.init_app(app)

    """
    LoginManager settings
    """
    login_manager.init_app(app)
    login_manager.login_message = "Lutfen uye olunuz."
    login_manager.login_view = "uye.giris"

    migrate = Migrate(app, db)
    Bootstrap(app)

    from app import models

    from .admin import admin as admin_blueprint
    app.register_blueprint(admin_blueprint, url_prefix='/admin')

    from .Uye import uye as uye_blueprint
    app.register_blueprint(uye_blueprint)

    from .home import home as home_blueprint
    app.register_blueprint(home_blueprint)

    @app.errorhandler(403)
    def forbidden(error):
        return render_template('errors/Yasak.html', title='Yasak'), 403

    @app.errorhandler(404)
    def page_not_found(error):
        return render_template('errors/404.html', title='Sayfa Bulunamadi'), 404

    @app.errorhandler(500)
    def internal_server_error(error):
        return render_template('errors/500.html', title='Server Hatasi'), 500


    return app

def create_app(taskflow_instance, connection_string=None, secret_key=None):
    app = flask.Flask(__name__)
    app.config['DEBUG'] = os.getenv('DEBUG', False)
    app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
    app.config['SQLALCHEMY_DATABASE_URI'] = connection_string or os.getenv('SQL_ALCHEMY_CONNECTION')
    app.config['SESSION_COOKIE_NAME'] = 'taskflowsession'
    app.config['SESSION_COOKIE_HTTPONLY'] = True
    app.config['PERMANENT_SESSION_LIFETIME'] = 43200
    app.config['SECRET_KEY'] = secret_key or os.getenv('FLASK_SESSION_SECRET_KEY')

    db = SQLAlchemy(metadata=metadata, model_class=BaseModel)

    db.init_app(app)
    api = Api(app)
    CORS(app, supports_credentials=True)

    login_manager = LoginManager()
    login_manager.init_app(app)

    @login_manager.user_loader
    def load_user(user_id):
        return db.session.query(User).filter(User.id == user_id).first()

    def apply_attrs(class_def, attrs):
        for key, value in attrs.items():
            setattr(class_def, key, value)
        return class_def

    attrs = {
        'session': db.session,
        'taskflow': taskflow_instance
    }

    with app.app_context():
        api.add_resource(apply_attrs(resources.LocalSessionResource, attrs), '/v1/session')

        api.add_resource(apply_attrs(resources.WorkflowListResource, attrs), '/v1/workflows')
        api.add_resource(apply_attrs(resources.WorkflowResource, attrs), '/v1/workflows/<workflow_name>')

        api.add_resource(apply_attrs(resources.TaskListResource, attrs), '/v1/tasks')
        api.add_resource(apply_attrs(resources.TaskResource, attrs), '/v1/tasks/<task_name>')

        api.add_resource(apply_attrs(resources.WorkflowInstanceListResource, attrs), '/v1/workflow-instances')
        api.add_resource(apply_attrs(resources.WorkflowInstanceResource, attrs), '/v1/workflow-instances/<int:instance_id>')

        api.add_resource(apply_attrs(resources.RecurringWorkflowLastestResource, attrs), '/v1/workflow-instances/recurring-latest')

        api.add_resource(apply_attrs(resources.TaskInstanceListResource, attrs), '/v1/task-instances')
        api.add_resource(apply_attrs(resources.TaskInstanceResource, attrs), '/v1/task-instances/<int:instance_id>')

        api.add_resource(apply_attrs(resources.RecurringTaskLastestResource, attrs), '/v1/task-instances/recurring-latest')

    return app