我们从Python开源项目中,提取了以下23个代码示例,用于说明如何使用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_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