def get_repo_if_admin(db, full_name):
"""Retrieve repository from db and return if
current user is admin (owner or member)
:param db: database connection where are repos stored
:type db: ``flask_sqlalchemy.SQLAlchemy``
:param full_name: full name of desired repository
:type full_name: str
:return: repository if found, None otherwise
:rtype: ``repocribro.models.Repository`` or None
"""
user = flask_login.current_user.github_user
repo = db.session.query(Repository).filter_by(
full_name=full_name
).first()
if repo is None:
return None
if repo.owner == user or user in repo.members:
return repo
return None
python类session()的实例源码
def organizations():
"""List user organizations from GitHub (GET handler)"""
page = int(flask.request.args.get('page', 0))
gh_api = flask.current_app.container.get(
'gh_api', token=flask.session['github_token']
)
gh_orgs = gh_api.get('/user/orgs', page=page)
orgs_link = gh_api.app_connections_link
return flask.render_template(
'manage/orgs.html', orgs=gh_orgs.data,
actual_page=gh_orgs.actual_page, total_pages=gh_orgs.total_pages,
orgs_link=orgs_link,
)
def organization(login):
"""List organization repositories for activation
.. :todo: register organization in repocribro
.. :todo: own profile page of organization
"""
ORG_REPOS_URL = '/orgs/{}/repos?type=member'
page = int(flask.request.args.get('page', 0))
gh_api = flask.current_app.container.get(
'gh_api', token=flask.session['github_token']
)
gh_repos = gh_api.get(ORG_REPOS_URL.format(login), page=page)
user = flask_login.current_user.github_user
active_ids = [repo.github_id for repo in user.repositories]
return flask.render_template(
'manage/repos.html', repos=gh_repos.data,
actual_page=gh_repos.actual_page, total_pages=gh_repos.total_pages,
Repository=Repository, active_ids=active_ids,
repos_type=login+' (organization)'
)
def github_callback_get_account(db, gh_api):
"""Processing GitHub callback action
:param db: Database for storing GitHub user info
:type db: ``flask_sqlalchemy.SQLAlchemy``
:param gh_api: GitHub API client ready for the communication
:type gh_api: ``repocribro.github.GitHubAPI``
:return: User account and flag if it's new one
:rtype: tuple of ``repocribro.models.UserAccount``, bool
"""
user_data = gh_api.get('/user').data
gh_user = db.session.query(User).filter(
User.github_id == user_data['id']
).first()
is_new = False
if gh_user is None:
user_account = UserAccount()
db.session.add(user_account)
gh_user = User.create_from_dict(user_data, user_account)
db.session.add(gh_user)
db.session.commit()
is_new = True
return gh_user.user_account, is_new
def gh_event_push(db, repo, payload, actor):
"""Process GitHub PushEvent (with commits)
https://developer.github.com/v3/activity/events/types/#pushevent
:param db: Database to store push data
:type db: ``flask_sqlalchemy.SQLAlchemy``
:param repo: Repository where push belongs to
:type repo: ``repocribro.models.Repository``
:param payload: Data about push and commits
:type payload: dict
:param actor: Actor doing the event
:type actor: dict
"""
push = Push.create_from_dict(payload, actor, repo)
db.session.add(push)
for commit in push.commits:
db.session.add(commit)
def gh_event_release(db, repo, payload, actor):
"""Process GitHub ReleaseEvent (with commits)
https://developer.github.com/v3/activity/events/types/#releaseevent
:param db: Database to store push data
:type db: ``flask_sqlalchemy.SQLAlchemy``
:param repo: Repository where release belongs to
:type repo: ``repocribro.models.Repository``
:param payload: Data about release and action
:type payload: dict
:param actor: Actor doing the event
:type actor: dict
"""
action = payload['action']
release = Release.create_from_dict(payload['release'], actor, repo)
db.session.add(release)
def make_githup_api_factory(cfg):
"""Simple factory for making the GitHub API client factory
:param cfg: Configuration of the application
:type cfg: ``configparser.ConfigParser``
:return: GitHub API client factory
:rtype: ``function``
"""
def github_api_factory(token=None, session=None):
return GitHubAPI(
cfg.get('github', 'client_id'),
cfg.get('github', 'client_secret'),
cfg.get('github', 'webhooks_secret'),
session=session,
token=token
)
return github_api_factory
def session(db, request):
"""Creates a new database session for a test."""
connection = db.engine.connect()
transaction = connection.begin()
options = dict(bind=connection, binds={})
session = db.create_scoped_session(options=options)
db.session = session
def teardown():
transaction.rollback()
connection.close()
session.remove()
request.addfinalizer(teardown)
return session
def test_session_transactions(self):
app = flask.Flask(__name__)
app.testing = True
app.secret_key = 'testing'
@app.route('/')
def index():
return text_type(flask.session['foo'])
with app.test_client() as c:
with c.session_transaction() as sess:
self.assert_equal(len(sess), 0)
sess['foo'] = [42]
self.assert_equal(len(sess), 1)
rv = c.get('/')
self.assert_equal(rv.data, b'[42]')
with c.session_transaction() as sess:
self.assert_equal(len(sess), 1)
self.assert_equal(sess['foo'], [42])
def test_session_using_application_root(self):
class PrefixPathMiddleware(object):
def __init__(self, app, prefix):
self.app = app
self.prefix = prefix
def __call__(self, environ, start_response):
environ['SCRIPT_NAME'] = self.prefix
return self.app(environ, start_response)
app = flask.Flask(__name__)
app.wsgi_app = PrefixPathMiddleware(app.wsgi_app, '/bar')
app.config.update(
SECRET_KEY='foo',
APPLICATION_ROOT='/bar'
)
@app.route('/')
def index():
flask.session['testing'] = 42
return 'Hello World'
rv = app.test_client().get('/', 'http://example.com:8080/')
self.assert_in('path=/bar', rv.headers['set-cookie'].lower())
def test_session_using_session_settings(self):
app = flask.Flask(__name__)
app.config.update(
SECRET_KEY='foo',
SERVER_NAME='www.example.com:8080',
APPLICATION_ROOT='/test',
SESSION_COOKIE_DOMAIN='.example.com',
SESSION_COOKIE_HTTPONLY=False,
SESSION_COOKIE_SECURE=True,
SESSION_COOKIE_PATH='/'
)
@app.route('/')
def index():
flask.session['testing'] = 42
return 'Hello World'
rv = app.test_client().get('/', 'http://www.example.com:8080/test/')
cookie = rv.headers['set-cookie'].lower()
self.assert_in('domain=.example.com', cookie)
self.assert_in('path=/', cookie)
self.assert_in('secure', cookie)
self.assert_not_in('httponly', cookie)
def login_required(func):
"""
Decorator to require login and save URL for redirecting user after login
"""
@wraps(func)
def decorated_function(*args, **kwargs):
"""decorator args"""
if not is_logged_in():
# Save off the page so we can redirect them to what they were
# trying to view after logging in.
session['previously_requested_page'] = request.url
return redirect(url_for('login'))
return func(*args, **kwargs)
return decorated_function
def results():
respondent = session['respondent']
print("Petitioner = %s" % session['petitioner'])
petitioner = session['petitioner']
return render_template('results.html',
title='Court Form Sample',
petitioner=petitioner,
respondent=respondent
)
#Error Handling:
# http://blog.miguelgrinberg.com/post/the-flask-mega-tutorial-part-vii-unit-testing
def test_session_transactions(self):
app = flask.Flask(__name__)
app.testing = True
app.secret_key = 'testing'
@app.route('/')
def index():
return text_type(flask.session['foo'])
with app.test_client() as c:
with c.session_transaction() as sess:
self.assert_equal(len(sess), 0)
sess['foo'] = [42]
self.assert_equal(len(sess), 1)
rv = c.get('/')
self.assert_equal(rv.data, b'[42]')
with c.session_transaction() as sess:
self.assert_equal(len(sess), 1)
self.assert_equal(sess['foo'], [42])
def test_session_using_application_root(self):
class PrefixPathMiddleware(object):
def __init__(self, app, prefix):
self.app = app
self.prefix = prefix
def __call__(self, environ, start_response):
environ['SCRIPT_NAME'] = self.prefix
return self.app(environ, start_response)
app = flask.Flask(__name__)
app.wsgi_app = PrefixPathMiddleware(app.wsgi_app, '/bar')
app.config.update(
SECRET_KEY='foo',
APPLICATION_ROOT='/bar'
)
@app.route('/')
def index():
flask.session['testing'] = 42
return 'Hello World'
rv = app.test_client().get('/', 'http://example.com:8080/')
self.assert_in('path=/bar', rv.headers['set-cookie'].lower())
def test_session_using_session_settings(self):
app = flask.Flask(__name__)
app.config.update(
SECRET_KEY='foo',
SERVER_NAME='www.example.com:8080',
APPLICATION_ROOT='/test',
SESSION_COOKIE_DOMAIN='.example.com',
SESSION_COOKIE_HTTPONLY=False,
SESSION_COOKIE_SECURE=True,
SESSION_COOKIE_PATH='/'
)
@app.route('/')
def index():
flask.session['testing'] = 42
return 'Hello World'
rv = app.test_client().get('/', 'http://www.example.com:8080/test/')
cookie = rv.headers['set-cookie'].lower()
self.assert_in('domain=.example.com', cookie)
self.assert_in('path=/', cookie)
self.assert_in('secure', cookie)
self.assert_not_in('httponly', cookie)
def get_google_authorization_url():
current_user = flask.g.user
if current_user.is_authenticated:
return
google = get_google_auth()
auth_url, state = google.authorization_url(Auth.AUTH_URI)
flask.session['oauth_state'] = state
return auth_url
def is_authenticated():
if "username" in session:
return True
else:
return False
def is_admin():
if not "username" in session:
return False
if not (session['usergroup'] == 'root' or session['usergroup'] == 'admin'):
return False
return True
def is_activated():
if not "username" in session:
return False
if not (session['status']=='normal'):
return False
return True
