lancium.api.Data

Data

This is an object that represents the Data area on Lancium Compute.

class Data:
Data(path=None, is_dir=False, **kwargs)

ARGS:

  • path (string): path in the persistent storage area
  • is_dir (bool): Boolean for whether data object is a directory
  • length (int): length of Data object in bytes
  • last_modified (string): last modified date and time of Data object
  • date_created (string): date and time of Data object creation
  • __key (string): auth key
def error_handle( res, path=None, byte_range_start=0, byte_range_end=-1, url=None, source_type=None, source=None, code=None, error_handling=True, key=None, recurse_count=0):

Helper Method -- Handles status checking for server response objects. No need to call directly.

ARGS:

  • res (response object): response object from the server
  • path (string): path with the data area
  • byte_range_start (int, optional): byte range start of the expected data
  • byte_range_end (int, optional): byte range end of the expected data
  • url (string, optional): url to do an API call to the server with
  • source_type (string, optional): type in (file, directory, data, data_recursive, url)
  • source (string, optional): source to retrieve data from (name of file for source_type 'file')

RETURNS: Nothing unless the request is a status code 422 in which case it returns list with the server response and a boolean stating that it is a directory.

@staticmethod
def create(path, source_type, source=None, force=False, **kwargs):

Create a new data object.

POST /data/

Args:

  • path (string): path in Lancium’s persistent storage area
  • source_type (string): type of Data (‘file’, ‘url’)
  • source (string): local path (source_type: file, url (source_type: url)
  • force (bool): forces data upload even if path already exists within the persistent storage
  • **kwargs(dictionary): can contain auth key if you would like to perform this method using a different account. {'auth': ANOTHER_API_KEY}

Returns: Data (lancium.api.Data): Data object

import os
from lancium.api.Data import Data

def fake_callback(total_chunks, current_chunk):
    pass

data = Data().create('test1', 'file', source = os.path.abspath('impish.simg'), force=True)
data.upload( os.path.abspath('impish.simg'),fake_callback)
print(data.__dict__)

###OUTPUT BELOW
{'_Data__key': None, 'path': 'test1', 'is_dir': False, 'length': None, 'last_modified': None, 'date_created': None}
def refresh(self):

Refresh metadata about a specific file or get contents of a directory.

HEAD /data/ if file

GET /data/ if directory

Args: None

Returns: None (None): None

import os
from lancium.api.Data import Data

def fake_callback(total_chunks, current_chunk):
    pass

data = Data.create('testing_path', 'file', os.path.abspath('asd.py'), True)
data.upload('asd.py', fake_callback)
print(data.is_dir)
data.is_dir = True
print(data.is_dir)
data.refresh()
print(data.is_dir)

###OUTPUT BELOW
False
True
False
@staticmethod
def get( path, byte_range_start=0, byte_range_end=-1, download_path=None, callback=None, **kwargs):

Get a Data object by path.

    **GET /data/<path>**

    Args:
    * `path (string)`: path in Lancium’s persistent storage area
    * `byte_range_start (int, optional)`: starting byte for Data if downloading data file
    * `byte_range_end (int, optional)`: ending byte for Data if downloading data file
    * `download_path (string, optional)`: location to save data from the persistent storage area
    * `callback (func, optional)`: called after each chunk is downloaded, accepts arguments in the form of (download stage, percent complete, current_chunk_contents)
    * `**kwargs (dictionary)`: can contain auth key if you would like to perform this method using a different account. {'auth': ANOTHER_API_KEY}
    Returns:
        Data (lancium.api.Data): a Data object

        ### PRINT IT OUT
        import os
        from time import sleep
        from lancium.api.Data import Data


        def desk(stage, progress, contents): 
            if contents != None:
                print(type(contents))
                contents = str(contents)
                contents = contents [2:-1]
                print(contents)

        def fake_callback(total_chunks, current_chunk):
            pass

        data = Data().create('test1', 'file', source = os.path.abspath('impish.simg'), force=True)
        data.upload(os.path.abspath('impish.simg'),fake_callback)
        sleep(5)
        ex1 = Data.get('test1', callback = desk)
        ### OUTPUT BELOW FROM ABOVE CODE

        <class 'bytes'>
        'Bootstrap: docker
From: ubuntu:bionic

%runscript
apt-get update 
'

        ### DOWNLOAD IT (NEW CODE)
        import os
        from time import sleep
        from lancium.api.Data import Data


        def fake_callback(total_chunks, current_chunk):
            pass

        data = Data().create('test1', 'file', source = os.path.abspath('impish.simg'), force=True)
        data.upload(os.path.abspath('impish.simg'),fake_callback)
        sleep(5)
        ex1 = Data.get('test1', download_path='./')
        os.system('ls test1')   

        ###OUTPUT BELOW
        test1
@staticmethod
def show(path, **kwargs):

Retrieve metadata about a specific file or get contents of a directory.

HEAD /data/ if file

GET /data/ if directory

Args:

  • path (string): path within the persistent data area.
  • **kwargs (dictionary): can contain auth key to perform this method using a different account. {'auth': ANOTHER_API_KEY}

Returns: list of Data objects: a list of Data of objects

import os
from lancium.api.Data import Data

def fake_callback(total_chunks, current_chunk):
    pass

Data().create('test1', 'file', source = os.path.abspath('impish.simg'), force=True)
Data().upload('test1', os.path.abspath('impish.simg'),fake_callback)

ex1 = Data.show('test1')[0]

print(ex1)

### OUTPUT BELOW
{'_Data__key': None, 'path': 'test1', 'is_dir': False, 'length': '66', 'last_modified': 'Thu, 13 Oct 2022 18:39:42 +0000', 'date_created': 'Thu, 13 Oct 2022 18:39:42 +0000'}
def children(self):

Retrieves the file / directory children of a directory object

HEAD data//

GET data//

Args: None

Returns: (list of lancium.api.Data): array of Data objects

from lancium.api.Data import Data

folder = Data.get('test2')

print(f"folder is Directory? {folder.is_dir}")

children = folder.children()

print(f"this is folder's children: {children}")

###OUTPUT BELOW
folder is Directory? True
this is folder's children: [<lancium.api.Data.Data object at 0x7fd0274591b0>, ... ,<lancium.api.Data.Data object at 0x7fd027b37d60>]
def flatten_children_to_arr(path, contents):

Flattens data objects to arrays.

Args:

  • contents (json): contents of the response json object.

Returns: array of Data objects

def upload(self, file_path, callback=None):

Upload file.

PATCH /data/

Args:

  • file_path (string): local file path of input file
  • callback (func, optional): called after each chunk is downloaded, accepts arguments in the form of (file_size, file_start, total_chunks, current_chunk)

Returns: response (Requests.response): response object in json format

import os
from lancium.api.Data import Data

def fake_callback(total_chunks, current_chunk):
    pass

data = Data().create('test1', 'file', source = os.path.abspath('impish.simg'), force=True)
data.upload(os.path.abspath('impish.simg'),fake_callback)

ex = data.show('test1')[0]
print(ex.__dict__)

###OUTPUT BELOW
{'_Data__key': None, 'path': 'test1', 'is_dir': False, 'length': '66', 'last_modified': 'Thu, 13 Oct 2022 18:41:00 +0000', 'date_created': 'Thu, 13 Oct 2022 18:41:00 +0000'}
@staticmethod
def delete(path, recursive=False, **kwargs):

Delete data object (file or directory)

DELETE /data/

ARGS:

  • path (string): path in Data area
  • recursive (boolean): flag for whether to delete recursive (directory)
  • **kwargs (dictionary): contains key 'auth' optionally to state which api key to use

RETURNS: None: None

from lancium.api.Data import Data


Data().delete('test1')
Data().show('test1')

###OUTPUT BELOW
The requested resource could not be found.
def destroy(self, recursive=False):

Delete data object.

DELETE /data/

ARGS:

  • recursive (boolean): flag for whether to delete recursive (directory)

RETURNS: None: None

from lancium.api.Data import Data

folder = Data.get('test2')

print(folder.path != None)

folder.destroy(recursive=True)

print(folder.path != None)

###OUTPUT BELOW
True
False