bversion/server.py
def pull_file(request: Request, context: ConnectionContext) -> Responce: # pylint: disable=W0613
    """ Get a file from the server """

    session_token = request.headers['session_token'].encode('utf8')
    repository    = request.headers['repository']

    #===
    current_user = have_authenticated_user(request.remote_addr, repository, session_token)
    if current_user is False: return fail(user_auth_fail_msg)


    #===
    data_store = versioned_storage(config['repositories'][repository]['path'],
                                   path_override = config['repositories'][repository]['transient_db_path_override'])

    use_head = bool(int(request.headers['use_head']))

    if use_head:
        version_id = data_store.get_head()
    else:
        version_id = request.headers['version_id']


    # =============
    file_info = data_store.get_file_info_from_path(request.headers['path'], version_id = version_id)

    full_file_path: str = cpjoin(data_store.get_file_directory_path(file_info['hash']), file_info['hash'][2:])
    return success({'file_info_json' : json.dumps(file_info)}, ServeFile(full_file_path))
 Get a file from the server 
Get a file from the server
Get a file from the server
def pull_file(request: Request, context: ConnectionContext) -> Responce: 
    session_token = request.headers['session_token'].encode('utf8')
    repository    = request.headers['repository']
    current_user = have_authenticated_user(request.remote_addr, repository, session_token)
    if current_user is False: return fail(user_auth_fail_msg)
    data_store = versioned_storage(config['repositories'][repository]['path'],
                                   path_override = config['repositories'][repository]['transient_db_path_override'])
    use_head = bool(int(request.headers['use_head']))
    if use_head:
        version_id = data_store.get_head()
    else:
        version_id = request.headers['version_id']
    file_info = data_store.get_file_info_from_path(request.headers['path'], version_id = version_id)
    full_file_path: str = cpjoin(data_store.get_file_directory_path(file_info['hash']), file_info['hash'][2:])
    return success({'file_info_json' : json.dumps(file_info)}, ServeFile(full_file_path))
def pull_file ( request : Request , context : ConnectionContext ) -> Responce : session_token = request . headers [ 'session_token' ] . encode ( 'utf8' ) repository = request . headers [ 'repository' ] current_user = have_authenticated_user ( request . remote_addr , repository , session_token ) if current_user is False : return fail ( user_auth_fail_msg ) data_store = versioned_storage ( config [ 'repositories' ] [ repository ] [ 'path' ] , path_override = config [ 'repositories' ] [ repository ] [ 'transient_db_path_override' ] ) use_head = bool ( int ( request . headers [ 'use_head' ] ) ) if use_head : version_id = data_store . get_head ( ) else : version_id = request . headers [ 'version_id' ] file_info = data_store . get_file_info_from_path ( request . headers [ 'path' ] , version_id = version_id ) full_file_path : str = cpjoin ( data_store . get_file_directory_path ( file_info [ 'hash' ] ) , file_info [ 'hash' ] [ 2 : ] ) return success ( { 'file_info_json' : json . dumps ( file_info ) } , ServeFile ( full_file_path ) )
Get a file from the server
Get a file from the server
# pylint: disable=W0613 """ Get a file from the server """ #=== #=== # =============
request
Request
context
ConnectionContext