Python app.models 模块，Role() 实例源码

def validate(self):
        #check for old pw hash and upadte password if needed
        self.user = db.session.query(models.User).filter(models.User.email == self.email.data).first()
        if self.user and self.user.password.startswith("pbkdf2:sha1"):
            if check_password_hash(self.user.password, self.password.data):
                self.user.password = encrypt_password(self.password.data)
                self.user.active = 1
                self.user.roles.append(db.session.query(models.Role).filter(models.Role.name=="admin").first())
                db.session.commit()
                return True

        #do the flask-security checks
        if not super(Login, self).validate():
            return False

        return True

def post(self):
        """ Assign a scheduling task """
        s = Schedule2.query\
            .join(Role)\
            .join(Location)\
            .join(Organization)\
            .filter(Schedule2.state == "mobius-queue", Organization.active == True)\
            .order_by(asc(Schedule2.last_update))\
            .first()

        if s is None:
            abort(404)

        s.transition_to_mobius_processing()

        role = Role.query.get(s.role_id)
        loc = Location.query.get(role.location_id)

        return {
            "schedule_id": s.id,
            "role_id": role.id,
            "location_id": loc.id,
            "organization_id": loc.organization_id,
        }

def get(self):
        """ Return all queued calculations """
        response = {}
        response[API_ENVELOPE] = {}

        for state in ['chomp-queue', 'chomp-processing']:
            data = Schedule2.query.join(Role)\
                .join(Location)\
                .join(Organization)\
                .filter(Schedule2.state == state, Organization.active == True)\
                .all()
            response[API_ENVELOPE][state] = map(
                lambda schedule: marshal(schedule, tasking_schedule_fields),
                data)

        return response

def post(self):
        """ Assign a scheduling task """
        s = Schedule2.query\
            .join(Role)\
            .join(Location)\
            .join(Organization)\
            .filter(Schedule2.state == "chomp-queue", Organization.active == True)\
            .order_by(asc(Schedule2.last_update))\
            .first()

        if s is None:
            abort(404)

        s.transition_to_chomp_processing()

        role = Role.query.get(s.role_id)
        loc = Location.query.get(role.location_id)

        return {
            "schedule_id": s.id,
            "role_id": role.id,
            "location_id": loc.id,
            "organization_id": loc.organization_id,
        }

def create_user(
        name=None, email=None,
        password=None, role=None, app=None):
    "Create a user"

    role, created = get_or_create(Role, name=role)

    with app.app_context():
        if all([name, email, password]):
            user = User(
                first_name=name,
                roles=[role],
                active=True,
                email=email
            )
            user.set_password(password)
            db.session.add(user)
        else:
            user = "Cant create the user"
        db.session.commit()
        print(user)

def configure_admin(app, db):
    admin.init_app(app)
    # Admin pages
    admin.add_view(
        BaseView(OperatingSystem, db.session, endpoint='operatingsystems'))
    admin.add_view(BaseView(Project, db.session, endpoint='projects'))
    admin.add_view(ReleaseView(Release, db.session, endpoint='releases'))
    admin.add_view(TestRunView(TestRun, db.session, endpoint='testruns'))

    # security
    admin.add_view(BaseView(User, db.session, endpoint='users',
                            category='Auth'))
    admin.add_view(BaseView(Role, db.session, endpoint='roles',
                            category='Auth'))

    @app.security.context_processor
    def security_context_processor():
        return dict(
            admin_base_template=admin.base_template,
            admin_view=admin.index_view,
            h=admin_helpers,
            get_url=url_for
        )

def test_delete_it_all(base):
    user1 = User.query.first(email=user_dict['email'])
    user2 = User.query.first(email=user_dict2['email'])
    role1 = Role.query.first(name='User')
    role2 = Role.query.first(name='Admin')
    assert user1.delete() is True
    assert role1.delete() is True
    assert user2.delete() is True
    assert role2.delete() is True
    user1 = User.query.first(email=user_dict['email'])
    user2 = User.query.first(email=user_dict2['email'])
    role1 = Role.query.first(name='User')
    role2 = Role.query.first(name='Admin')
    assert user1 is None
    assert role1 is None
    assert user2 is None
    assert role2 is None


####################################################################################################
# Ensure we've closed all the RethinkDB connections
####################################################################################################

def startScheduler():
    db.create_all()
    #create default roles!
    if not db.session.query(models.Role).filter(models.Role.name == "admin").first():
        admin_role = models.Role(name='admin', description='Administrator Role')
        user_role = models.Role(name='user', description='User Role')
        db.session.add(admin_role)
        db.session.add(user_role)
        db.session.commit()

    try:
        import tzlocal

        tz = tzlocal.get_localzone()
        logger.info("local timezone: %s" % tz)
    except:
        tz = None

    if not tz or tz.zone == "local":
        logger.error('Local timezone name could not be determined. Scheduler will display times in UTC for any log'
                 'messages. To resolve this set up /etc/timezone with correct time zone name.')
        tz = pytz.utc
    #in debug mode this is executed twice :(
    #DONT run flask in auto reload mode when testing this!
    scheduler = BackgroundScheduler(logger=sched_logger, timezone=tz)
    scheduler.add_job(notify.task, 'interval', seconds=config.SCAN_INTERVAL, max_instances=1,
                      start_date=datetime.datetime.now(tz) + datetime.timedelta(seconds=2))
    scheduler.start()
    sched = scheduler
    #notify.task()

def make_shell_context():
    """??????"""
    return dict(
        app = app,
        db = db,
        User = User,
        Role = Role,
        Comment = Comment
    )

def make_shell_context():
    return dict(app=app, db=db, User=User, Role=Role)

def make_shell_context():
    return dict(app=app, db=db, User=User, Role=Role)

def make_shell_context():
    return dict(app=app, db=db, User=User, Role=Role,Question=Question,Category=Category,Usertoca=Usertoca,Potoca=Potoca,Vote=Vote)

def make_shell_context():
    return dict(app=app, db=db, User=User, Role=Role, Tag=Tag, Video=Video, videotags=videotags)

def test_role_model(self):
        """
        Test number of records in Role table
        """

        # create test role
        role = Role(name="CEO", description="Run the whole company")

        # save role to database
        db.session.add(role)
        db.session.commit()

        self.assertEqual(Role.query.count(), 1)

def get(self):
        """ Return all queued calculations """
        response = {}
        response[API_ENVELOPE] = {}
        for state in ['mobius-queue', 'mobius-processing']:
            data = Schedule2.query\
                .join(Role)\
                .join(Location)\
                .join(Organization)\
                .filter(Schedule2.state == state, Organization.active == True)\
                .all()
            response[API_ENVELOPE][state] = map(
                lambda schedule: marshal(schedule, tasking_schedule_fields),
                data)
        return response

def make_shell_context():
    return dict(app=app, db=db, User=User, Role=Role, Permission=Permission)

def make_shell_context() :
    return dict(app=app , dn=db , User=User,Role=Role)

def make_shell_context():
    return dict(app=app, db=db, User=User, Role=Role, Permission=Permission,
                Post=Post)

def create_app():
    """
    App factory pattern avoids circular imports, so instead of importing
    'app' directly you import its factory. If you need the current running app
    you can use 'from flask import current_app'
    :return: app
    """
    app = Flask(__name__)
    config_name = os.environ.get('FLASK_CONFIG') or 'development'
    cfg = os.path.join(os.getcwd(), 'config', config_name + '.py')
    app.config.from_pyfile(cfg)

    # initialize extensions
    bootstrap.init_app(app)
    db.init_app(app)

    # security
    user_datastore = SQLAlchemyUserDatastore(db, User, Role)
    app.security = Security(app, user_datastore)

    # import blueprints
    from app.dashboard.views import dashboard_blueprint
    from app.operatingsystem.views import os_blueprint
    from app.project.views import project_blueprint
    from app.release.views import release_blueprint
    from app.testrun.views import run_blueprint

    app.register_blueprint(dashboard_blueprint)
    app.register_blueprint(os_blueprint)
    app.register_blueprint(project_blueprint)
    app.register_blueprint(release_blueprint)
    app.register_blueprint(run_blueprint)

    configure_admin(app, db)

    db.create_all(app=app)

    return app

def make_shell_context():
    return dict(app = app, db = db, User = User, Role = Role, Post = Post)

def set_role(id, role):
    user = User.query.get(id)
    role = Role.query.filter_by(name=role).first()
    if not user:
        print 'User not found'
        exit(1)
    if not role:
        print 'Role not found'
        exit(1)
    user.roles.append(role)
    db.session.add(user)
    db.session.commit()
    print 'Role set successfully.'

def unset_role(id, role):
    user = User.query.get(id)
    role = Role.query.filter_by(name=role).first()
    if not user:
        print 'User not found'
        exit(1)
    if not role:
        print 'Role not found'
        exit(1)
    user.roles.remove(role)
    db.session.add(user)
    db.session.commit()
    print 'Role unset successfully.'

def add_role(name):
    role = Role(name=name)
    db.session.add(role)
    db.session.commit()
    print '%s added' % unicode(role)

def delete_role(name):
    role = Role.query.filter_by(name=name).first()
    if not role:
        print 'Role not found'
        exit(1)
    db.session.delete(role)
    db.session.commit()
    print 'Role %s deleted' % name

def make_shell_context():
    return dict(app = app,db = db,User = User, Role = Role,Review =Review)

def make_shell_context():
    return dict(app=app,db=db,User=User,Role=Role,BopGame=BopGame,OwGame=OwGame,OwGlobal=OwGlobal,OwCross=OwCross)

def make_shell_context():
    return dict(app=app, db=db, User=User, Role=Role, Permission=Permission, Word=Word, Note=Note, UserWord=UserWord)

def make_shell_context():
    return dict(app=app, db=db, User=User, Follow=Follow, Role=Role, Permission=Permission, Post=Post, Comment=Comment)

def test_create_role(base):
    role = Role(role_dict)
    role.save()
    assert isinstance(role, Role)
    assert role.name == role_dict['name']
    assert role.description == role_dict['description']

def test_create_admin_role(base):
    role = Role(role_dict_admin)
    role.save()
    assert isinstance(role, Role)
    assert role.name == role_dict_admin['name']
    assert role.description == role_dict_admin['description']

def test_append_role(base):
    _u = User.query.first(email=user_dict['email'])
    assert _u is not None
    assert isinstance(_u, User)
    _r = Role.query.first(name=role_dict['name'])
    assert _r is not None
    assert isinstance(_r, Role)
    _u.roles.append(_r)
    _u.save()
    assert _r in _u.roles
    user = User.query.get(_u.id)
    assert isinstance(user, User)
    assert _r in user.roles

def test_mr_contains_role(base):
    assert 'Role' in model_registry._models

def make_shell_context():
    return dict(app = app, db = db, User = User, Role = Role, Permission = Permission)

def make_shell_context():
    return dict(app=app, db=db, User=User, Role=Role, Post=Post, Follow=Follow, Comment=Comment)

def test_role_model(self):
        """
        Test number of records in Role table
        """

        # create test role
        role = Role(name="CEO", description="Run the whole company")

        # save role to database
        db.session.add(role)
        db.session.commit()

        self.assertEqual(Role.query.count(), 1)

def setUp(self):
        """Setup the test driver and create test users"""
        self.driver = webdriver.Chrome()
        self.driver.get(self.get_server_url())

        db.session.commit()
        db.drop_all()
        db.create_all()

        # create test admin user
        self.admin = Employee(username=test_admin_username,
                              email=test_admin_email,
                              password=test_admin_password,
                              is_admin=True)

        # create test employee user
        self.employee = Employee(username=test_employee1_username,
                                 first_name=test_employee1_first_name,
                                 last_name=test_employee1_last_name,
                                 email=test_employee1_email,
                                 password=test_employee1_password)

        # create test department
        self.department = Department(name=test_department1_name,
                                     description=test_department1_description)

        # create test role
        self.role = Role(name=test_role1_name,
                         description=test_role1_description)

        # save users to database
        db.session.add(self.admin)
        db.session.add(self.employee)
        db.session.add(self.department)
        db.session.add(self.role)
        db.session.commit()

def make_shell_context():
    return dict(app=app,db=db,User=User,Role=Role,Permission=Permission,Post=Post,Follow=Follow)

def get(self, org_id, location_id):
        response = {
            API_ENVELOPE: [],
        }

        parser = reqparse.RequestParser()
        parser.add_argument("recurse", type=inputs.boolean, default=False)
        parser.add_argument("archived", type=inputs.boolean)
        args = parser.parse_args()
        args = dict((k, v) for k, v in args.iteritems() if v is not None)

        roles_query = Role.query.filter_by(location_id=location_id)

        # by default, only include active users
        if "archived" in args:
            roles_query = roles_query.filter_by(archived=args["archived"])

        roles = roles_query.all()
        response[API_ENVELOPE] = map(lambda role: marshal(role, role_fields),
                                     roles)

        if args["recurse"]:

            # Show all users in each location role
            for datum in response[API_ENVELOPE]:
                users_query = RoleToUser.query.filter_by(role_id=datum["id"])

                if "archived" in args:
                    users_query = users_query.filter_by(
                        archived=args["archived"])

                members = users_query.all()
                memberships = []

                for member in members:
                    rtu = marshal(member, role_to_user_fields)
                    rtu.update(marshal(member.user, user_fields))
                    memberships.append(rtu)

                datum.update({"users": memberships})

        return response