References

Module backo. export all classes

class backo.Action(schema: dict, on_trig: Callable, **kwargs)

Bases: Dict

The action to do on a collection

Parameters:

  • schema (dict) – The data schema needed for this action (see Dict <https://stricto.readthedocs.io/en/latest/api_reference.html#stricto.Dict>)

  • on_trig (Callable) – the function to trig on the action

  • **kwargs – see https://stricto.readthedocs.io/en/latest/api_reference.html#stricto.Dict - can_see= [func]|bool – a function to say if this action exists - can_execute= [func]|bool – a function to say if the CurrentUser can execute this action.

from backo import Item, Collection, Backoffice, Action, Ref, DBMongoConnector

# example
book_item = Item({
    "title": String(),
    "subtitle": String(),
    "author": Ref(coll="authors", field="$.books", required=True),
    "score": Float( default=5.0 )
    "number_of_voter" : Int(default=0)
})

def do_vote(action, o):
    o.score = (o.score * o.number_of_voter + action.score) / ( o.number_of_voter + 1 )
    o.number_of_voter += 1


vote_action = Action( {
    "score" : Float( max=10.0, min=0.0 )
}, do_vote)

database_for_books = DBMongoConnector( connection_string="mongodb://localhost:27017/bookcase" )
books = Collection( "books", book_item, database_for_books )
books.register_action( "vote", vote_action )

my_bookstore = Backoffice("bookstore")
my_bookstore.register_collection(books)
# ...
Parameters:

  • schema (dict) – The data schema needed for this action (see Dict <https://stricto.readthedocs.io/en/latest/api_reference.html#stricto.Dict>)

  • on_trig (Callable) – the function to trig on the action

  • **kwargs – see https://stricto.readthedocs.io/en/latest/api_reference.html#stricto.Dict - can_see= [func]|bool – a function to say if this action exists - can_execute= [func]|bool – a function to say if the CurrentUser can execute this action.

class backo.Backoffice(name: str)

Bases: object

Backoffice is tha application itself.

Parameters:

name (str) – A name for the application

from backo import Backoffice

# example
my_app = Backoffice("customer_app")

Constructor for backoffice

build_routes(flask_app: Flask, prefix: str = '', jwt_auth: Callable | None = None) → None

Add all routes to flask application

Parameters:

  • flask_app (Flask) – The Flask object

  • prefix (str) – an optional prefix to the path

  • jwt_auth – The token function

check_syntax() → None

Check the syntax by trigging a “check_syntax” event to all Items this event is catch by Ref and RefsList wich display warning if needed

collections: dict = {}

The list of collections, per name

migrate(collection_name: str, migration_function: Callable | None = None, **kwargs) → MigrationReport

_summary_

Parameters:

  • collection_name (str | list[str] | None) – _description_

  • _id (_type_) – _description_

Returns:

_description_

Return type:

_type_

name: str = None

The name of the backoffice

register_collection(coll: Collection) → None

Register a collection into this backoffice

Parameters:

coll (Collection) – The collection to add.

See Collection

class backo.Collection(name: str, model: Item, db_handler: DBConnector, **kwargs)

Bases: object

The Collection refer to a “table”

A collection is the main object in backo. It contains

  • an Item = the description of the object structure

  • an DBConnector = the database connector to say how and where to save the object

  • some Selection = some preset select for this collection

  • some Action = a list of actions to do on this collection

A collection mus by registered into a Backoffice with Backoffice.register_collection()

Parameters:

  • name – The collection name

  • model (Item) – The description of the structure (an Item)

  • db_handler (DBConnector) – The database handler

  • **kwargs –

    • refuse_filter= func – not used yet

    • can_read= [func]|bool – a function to say if the CurrentUser can read this collection

    • can_create= [func]|bool – a function to say if the CurrentUser can create an Item in this collection

    • can_delete= [func]|bool – a function to say if the CurrentUser can delete an Item in this collection

    • can_modify= [func]|bool – a function to say if the CurrentUser can modify an Item in this collection

from backo import Item, Collection, Backoffice, , DBMongoConnector

# example
book_item = Item({
    "title": String(),
    "subtitle": String(),
    "author": Ref(collection="authors", field="$.books", required=True),
})


database_for_books = DBMongoConnector( connection_string="mongodb://localhost:27017/bookcase" )
books = Collection( "books", book_item, database_for_books )

my_bookstore = Backoffice("bookstore")
my_bookstore.register_collection(books)
# ...

Constructor

add_action(name: str, action: Action)

See register_action()

create(obj: dict, **kwargs) → Item

Create and save an item into the DB

Parameters:

  • obj (dict) – The json object struture to create

  • **kwargs –

    • transaction_id= int – the current transaction_id (in case of rollback)

    • m_path= [str] – the modification path, to to avoid loop with references

Returns:

an empty Item

Return type:

Item

db_handler: DBConnector = None

The database connector

do_post_selection(_selection_name: str)

Do a selection with filter given in post

Parameters:

_selection_name (str) – The name of the selection

Raises:

Error – _description_

Returns:

_description_

Return type:

_type_

do_selection(_selection_name: str)

_summary_

Parameters:

_selection_name (str) – The name of the selection

Raises:

Error – _description_

Returns:

_description_

Return type:

_type_

get_by_id(_id: str) → Item

Return an object by Id.

Parameters:

_id (str) – the _id of the Item you want

Returns:

The item

Return type:

Item

get_meta() → dict

Return the meta data for this collection and actions

get_openapi_routes() → dict

Return the paths section of the OpenAPI specification.

get_openapi_schemas() → dict

Return the schemas section of the OpenAPI specification.

get_other_collection(name) → Self

Return another collection (used by Ref and RefsList)

Parameters:

name – the name of the collection you want

Returns:

the collection

Return type:

Self

migrate(migration_function: Callable | None, _ids: list[str] | None = None, dry_run: bool = True) → MigrationReport

start the migration for a collection

Parameters:

  • _ids (list[str] | None) – list of ids to migrate

  • dry_run (bool) – _description_just check

model: Item = None

The item definition

name: str = None

The name of the collection

new_item() → Item

Return an empty Item

Returns:

an empty Item

Return type:

Item

register_action(name: str, action: Action)

Register an action to this collection

Parameters:

  • name (str) – The name of the action

  • action (Action) – the Action to register

register_selection(selection_name: str, selection: Selection) → None

Register a selection to this collection

Parameters:

  • name (str) – The name of the selection

  • action (Selection) – the Selection to register

select(filter_for_selection: dict) → list[Item]

Do a selection directly with a filter

Parameters:

filter_for_selection (dict) – a filter

Returns:

a list of Items

Return type:

list[Item]

select_one(filter_for_selection: dict) → Item

select one item (if only one)

Parameters:

filter_for_selection (dict) – a filter

Returns:

The one Item matching the filter

Return type:

Item

class backo.CurrentUser(**kwargs)

Bases: Dict

the current connected user object

the current user is a Dict with

  • _id (String) : The _id of the user

  • login (String) : The login as a string

  • roles (List of String) : The list of roles for this user. A role is a string which can be used for rights.

Parameters:

**kwargs – see https://stricto.readthedocs.io/en/latest/api_reference.html#stricto.Dict

Constructor

has_role(role: str | list[str]) → bool

Return if the currentUser has the role given in param

Parameters:

role (str or list[str]) – the role

Returns:

a boolean if the user as the role has the role

Return type:

bool

# example
if current_user.has_role( "Admin" ):
    return True
else:
    return False
is_anonymous() → bool

Return True if the current_user is ANONYMOUS

Returns:

the result of the check anonymous

Return type:

bool

class backo.Item(schema: dict, **kwargs)

Bases: Dict

The description of the object of a collection

Parameters:
from backo import Item, Collection, Backoffice, DBMongoConnector

# example
book = Item({
    "title": String(),
    "subtitle": String(),
    "author": Ref(coll="authors", field="$.books", required=True),
    "pages": Int()
})

database_for_books = DBMongoConnector( connection_string="mongodb://localhost:27017/bookcase" )
books = Collection( book, database_for_books )

bookstore = Backoffice("bookstore")
bookstore.register_collection(books)
# ...

Constructor

create(obj: dict, **kwargs)

Create and save an object into the DB

Parameters:

  • obj (dict) – The json object struture to create

  • **kwargs –

    • transaction_id= int – the current transaction_id (in case of rollback)

    • m_path= [str] – the modification path, to to avoid loop with references

delete(**kwargs) → None

delete the object in the database

Parameters:

**kwargs –

  • transaction_id= int – the current transaction_id (in case of rollback)

  • m_path= [str] – the modification path, to to avoid loop with references

load(_id: str, **kwargs) → None

Read in the database by Id and fill the Data

Parameters:

  • _id (str) – The _id to load.

  • **kwargs –

    • transaction_id= int – the current transaction_id (in case of rollback)

    • m_path= [str] – the modification path, to to avoid loop with references

meta_data_handler: GenericMetaDataHandler = None

The metadata handler to apply

reload(**kwargs) → None

Reload from DB the object (in case of changement)

Parameters:

**kwargs –

  • transaction_id= int – the current transaction_id (in case of rollback)

  • m_path= [str] – the modification path, to to avoid loop with references

save(**kwargs) → None

save the object in the database.

Parameters:

**kwargs –

  • transaction_id= int – the current transaction_id (in case of rollback)

  • m_path= [str] – the modification path, to to avoid loop with references

class backo.Selection(selectors: list[str] | None = None, **kwargs)

Bases: CollectionAddon

The Selection refer to a select on a “table”

A collection must by registered into a Collection with Collection.register_selection()

Parameters:

  • selectors (list[str]) – The list of paths we went to see in the selection

  • **kwargs –

    • filter= dict|tuple – the filter whe want. See stricto for details

    • db_filter= dict – The filter to pass to the DBConnector

from backo import Item, Collection, Selection

# example
book_item = Item({
    "title": String(),
    "subtitle": String(),
    "author": Ref(collection="authors", field="$.books", required=True),
})

database_for_books = DBMongoConnector( connection_string="mongodb://localhost:27017/bookcase" )
books = Collection( "books", book_item, database_for_books )

fb = Selection( [ "$.title", "$.subtitle" ], filter={ "$.author.nationality.a2" : "FR" } )
books.register_selection("french_book", fb )

nfb = Selection( [ "$.title", "$.subtitle" ], filter={ "$.author.nationality.a2" : ( "$ne", "FR" ) } )
books.register_selection("non_french_book", nfb )

# ...

Selection constructor

Parameters:

**kwargs

  • filter= dict|tuple – the filter whe want. See stricto for details

  • db_filter= dict – The filter to pass to the DBConnector

can_read() → bool

return True if permission to read

get_schema() → dict

Return schema for this selection

Parameters:

self – Description

Returns:

the schema as a json object (dict)

Return type:

dict

Return the schema for this selection

select(match_filter=None, page_size=0, num_of_element_to_skip=0, db_sort_object={'_id': 1})

Do the selection

Ref & RefsList

class backo.Ref(**kwargs)

Bases: String

Ref 0 or 1 to many to another Collection

Parameters:

**kwargs –

  • collection|coll= str – The target collection

  • reverse|rev|field= str – The field in the target collection which reference my collection. Must be a RFC 9535 path (https://datatracker.ietf.org/doc/rfc9535/)

from backo import Item, Ref, RefsList

# example
book_item = Item({
    "title": String(),
    "author": Ref(collection="authors", field="$.books", required=True),
})
author_item = Item({
    "name": String(),
    "surname": String(),
    "books": RefsList(collection="books", field="$.author"),
})

books = Collection( "books", book_item, database_for_books )
authors = Collection( "authors", author_item, database_for_authors )

my_bookstore = Backoffice("bookstore")
my_bookstore.register_collection(books)
my_bookstore.register_collection(authors)

Constructor

check_syntax(event_name: str, root, me, **kwargs)

Check if everything is correct log some warnings

get_schema() → dict

get schema for ref with specific elements collection and reverse

Returns:

the schema

Return type:

dict

class backo.RefsList(**kwargs)

Bases: List

Ref 0 or 1 to many to another Collection

Parameters:

**kwargs – 

from backo import Item, Ref, RefsList, DeleteStrategy

# example
book_item = Item({
    "title": String(),
    "author": Ref(collection="authors", field="$.books", required=True),
})
author_item = Item({
    "name": String(),
    "surname": String(),
    "books": RefsList(collection="books", field="$.author", ods=DeleteStrategy.DELETE_REFERENCED_ITEMS),
})

books = Collection( "books", book_item, database_for_books )
authors = Collection( "authors", author_item, database_for_authors )

my_bookstore = Backoffice("bookstore")
my_bookstore.register_collection(books)
my_bookstore.register_collection(authors)

Constructor

check_syntax(event_name: str, root, me, **kwargs)

Check if everything is correct log some warnings

get_other_with_a_select(root_id: str) → list

Get reverse Items with a select (when FillStrategy.NO_FILL)

Returns:

list of Items

Return type:

list

get_schema() → dict

get schema for ref with specific elements collection and reverse

Returns:

the schema

Return type:

dict

class backo.DeleteStrategy(*values)

Bases: Enum

Specifics strategy for deletion for RefsList

when the user want to delete the object, if the object contains a RefsList. Say how to handle the deletion

  • MUST_BE_EMPTY = The RefsList must be empty otherwise the delete action will raise an error.

  • DELETE_REFERENCED_ITEMS = All objects targeted with this RefsList will be deleted too. Caution !

  • UNLINK_REFERENCED_ITEMS = The reverse field of all objects targeted with this RefsList will be cleaned

class backo.FillStrategy(*values)

Bases: Enum

Specifics strategy for fill RefsList in case of one_to_many or many_to_many links

  • FILL = The reverse is a List of _ids. Usefull to manage which is pointing to me.

  • NOT_FILL = Whe don’t want to fill because the list is to big (for example person -> nationality) but is important to keep the information of this link.

Files

class backo.File(**kwargs)

Bases: Dict

Object to manage files. inherite from https://stricto.readthedocs.io/en/latest/api_reference.html#stricto.Dict

Parameters:
check_type(value)

check if conplain to model or raise an

compute_value() → bool

compute the value if needed

Returns:

True if changed

Return type:

bool

copy_file_content(f_id_src: str, src: FileConnector, f_id_dst: str, dst: FileConnector) → None

Copy a file from an connector to another

Parameters:

  • f_id_src (str) – The source file_id

  • src (FileConnector) – The source FileConnector

  • f_id_dst (str) – The destination file_id

  • dst (FileConnector) – The destination FileConnector

delete_content() → None

Delete the file

generate_file_id() → str

The function to generate an Id.

Parameters:

o (dict) – The object given (json format)

Returns:

an Id

Return type:

str

get_content() → bytes | None

Get the file content

Returns:

the content of the file

Return type:

str|bytes

has_file() → bool

Check if the file exists / is set. (means it has a file_id)

Returns:

True if OK

Return type:

bool

load() → None

load from the cold storage

the cold storage must exists (params storage= in initialisation)

on_before_delete(event_name, root, me, **kwargs)

Try to the event “before_delete”

on_before_save(event_name, root, me, **kwargs)

Trigg by the event “before_save”

read_content_generator() → Generator

Return a generator which read chunk into the storage

Yield:

a chunk

Return type:

Generator

save() → None

Save to the cold storage

the cold storage must exists (params storage= in initialisation) This is called by save() from :py:class:Item before save the Item itself.

set_content(content: str | bytes | FileStorage) → None

Set the file content

set_value(value: Any) → bool

Set the value of the file.

If the value is a dict, handle it as file structure. if a str|bytes|FileStorage, set the content and fill meta datas in the structure.

Parameters:

value (Any) – The file Structure or content

Returns:

True if modifications or False if nothing changed.

Return type:

bool

class backo.BlobFile(**kwargs)

Bases: File

File type

Parameters:

**kwargs – See File

Object to manage files directly in the data File object.

With this object, file are store with meta_datas ( filename, file_id, size, content_type…) in the DB Interesting for small files.

copy_file_content(fsrc: str, src: FileConnector, fsdt: str, dst: FileConnector) → str | None

Copy a file from an connector to another

Parameters:
get_content() → bytes | None

Get the file content

Returns:

the content of the file

Return type:

str|bytes

load() → None

load from the cold storage

on_before_save(event_name, root, me, **kwargs)

Try to save the file if possible

read_chunk() → Generator

Read chunk by chunk

Yield:

a chunk

Return type:

Generator

read_content_generator() → Generator

Return a generator which read chunk into the storage

Yield:

a chunk

Return type:

Generator

save() → None

Save to the cold storage _storage_connector

Files storage

class backo.FileConnector(**kwargs)

Bases: ABC

Abstract Class fir File connector. this is the family (parent) object for files connector

Parameters:

  • kwargs (object) – arguments as kwargs for the FileConnector

  • buffer_size (int) – The biffer size (default = 0812)

abstractmethod clear(file_id: str) → None

Set the file given by its file_id to empty

Parameters:

file_id (str) – file id

abstractmethod delete(filename: str) → None

Delete the file

Parameters:

file_id (str) – file id

generate_id() → str

The function to generate a random id.

Returns:

an Id

Return type:

str

abstractmethod get(file_id: str, mode: str = 'rb', encoding: str | None = None) → bytes

Return the content of the file

Parameters:

file_id (str) – file id

Returns:

the file content

Return type:

bytes

abstractmethod has_file(file_id: str) → bool

check if the file exists

Parameters:

file_id (str) – file id

Returns:

True if the file exists

Return type:

bool

abstractmethod read_chunk(file_id: str) → Generator

Read a chunk of the file

Parameters:

file_id (str) – file id

Returns:

a generator to get the next chunk

Return type:

Generator

abstractmethod write_chunk(filename: str, chunk: bytes) → str | None

Append a chunk to the file

Parameters:

  • file_id (str) – file id

  • chunk (bytes) – the chunk content

class backo.FileSystemConnector(**kwargs)

Bases: FileConnector

File connector on a file system

This is the way to save / store / retrieve objects

Parameters:

  • kwargs (object) – arguments as kwargs for the FileSystemConnector

  • path (str) – The path to store files

clear(file_id: str) → None

Set the file given by its file_id to empty

Parameters:

file_id (str) – file id

delete(file_id: str) → None

Delete the file on dist

Parameters:

file_id (str) – file id

get(file_id: str, mode: str = 'rb', encoding: str | None = None) → bytes | str

Return the content of the file

Parameters:

file_id (str) – file id

Returns:

the file content

Return type:

bytes

has_file(file_id: str) → bool

check if the file exists

Parameters:

file_id (str) – file id

Returns:

True if the file exists

Return type:

bool

read_chunk(file_id: str) → Generator

Read a chunk of the file

Parameters:

file_id (str) – file id

Returns:

a generator to get the next chunk

Return type:

Generator

write_chunk(file_id: str, chunk: bytes) → None

Append a chunk to the file

Parameters:

  • file_id (str) – file id

  • chunk (bytes) – the chunk content

class backo.FileBlobConnector(**kwargs)

Bases: FileConnector

File connector in memory This is the way to save files just in memory.

clear(file_id: str) → None

Set the file given by its file_id to empty

Parameters:

file_id (str) – file id

delete(file_id: str) → None

Clear

Parameters:

file_id (str) – _description_

get(file_id: str, mode: str = 'rb', encoding: str | None = None) → bytes | str

Return the content of the file

Parameters:

file_id (str) – file id

Returns:

the file content

Return type:

bytes

has_file(file_id: str) → bool

check if the file exists

Parameters:

file_id (str) – file id

Returns:

True if the file exists

Return type:

bool

read_chunk(file_id: str) → Generator

Read a chunk of the file

Parameters:

file_id (str) – file id

Returns:

a generator to get the next chunk

Return type:

Generator

write_chunk(file_id: str, chunk: bytes) → None

Append a chunk to the file

Parameters:

  • file_id (str) – file id

  • chunk (bytes) – the chunk content

Database

class backo.DBConnector(**kwargs)

Bases: ABC

Database Connector

This is the way to save / store / retrieve objects

Parameters:

**kwargs –

  • restriction= func – not used yet

Constructor

abstractmethod create(o: dict) → str

Create the object into the DB and return the _id

Parameters:

o (dict) – The object given (json format)

Raises:

Error – Raise an error DBError or any db error

abstractmethod delete_by_id(_id: str)

The _id to delete on the db

Parameters:

_id (str) – the _id

Raises:

Error – Raise an error DBError or any db error

abstractmethod drop()

Drop the collection

Mainly used in test

Raises:

Error – Raise an error DBError or any db error

generate_id(o: dict) → str

The function to generate an Id.

Mainly, not used, because the database itself do the job (like mongo). But for other cases, you must generate by yourself the uniq _id for the object

Parameters:

o (dict) – The object given (json format)

Returns:

an Id

Return type:

str

abstractmethod get_by_id(_id: str) → dict

get an object by _id in the DB and return it

Parameters:

_id (str) – the _id

Returns:

The object (json format)

Return type:

dict

Raises:

Error – Raise an error DBError or any db error

abstractmethod save(_id: str, o: dict)

Save the objet

Parameters:

  • _id (str) – the _id of this object

  • o (dict) – The object given (json format)

Raises:

Error – Raise an error DBError or any db error

abstractmethod select(select_filter, projection: dict = {}, page_size: int = 0, num_of_element_to_skip: int = 0, sort_object: dict = {})

Select from filter in the DB and return a list of dicts, with pagination

Parameters:

  • select_filter – The filter for selection (depends on DB types)

  • projection (dict) – The list of elements we want for each object

  • page_size (int) – number of elements per page

  • num_of_element_to_skip (int) – number of element to skip from beginning

  • sort_object (dict) – Soon

Raises:

Error – Raise an error DBError or any db error

class backo.DBMongoConnector(**kwargs)

Bases: DBConnector

Mongodb database Connector

This is the way to save / store / retrieve objects in a mongodb

Parameters:

**kwargs –

  • restriction= func – not used yet

  • all other params are passed to Mongoclient

constructor

close()

Close the mongodb connection

Raises:

DBError – Raise an error in case of database Error

connect()

Try to make a connection to the mongodb

Raises:

BDError – Raise an error in case of database Error

create(o: dict)

See DBConnector.create()

delete_by_id(_id: str)

See DBConnector.delete_by_id()

drop()

See DBConnector.drop()

Raises:

DBError – Raise an error in case of database Error

generate_id(o)

Do not create _id by ourself. mongo will do the job

get_by_id(_id: str)

See DBConnector.get_by_id()

save(_id: str, o: dict)

See DBConnector.save()

select(select_filter, projection={}, page_size=0, num_of_element_to_skip=0, sort_object={'_id': 1})

See DBConnector.select()

Parameters:

select_filter (dict ( a mongodb fliter syntax )) – The filter for selection

class backo.DBYmlConnector(**kwargs)

Bases: DBConnector

Yaml files database Connector

This is the way to save / store / retrieve objects in yaml files

Parameters:

**kwargs –

  • path= str – The directory to store yaml files

constructor

create(o: dict) → str

See DBConnector.create()

delete_by_id(_id: str) → bool

See DBConnector.delete_by_id()

drop() → None

See DBConnector.drop()

get_by_id(_id: str) → dict

See DBConnector.get_by_id()

save(_id: str, o: dict) → None

See DBConnector.save()

select(select_filter, projection={}, page_size=0, num_of_element_to_skip=0, sort_object={'_id': 1}) → list

See DBConnector.select()

Params select_filter and projection are not used

class backo.DBRestfullConnector(**kwargs)

Bases: DBConnector

DBConnector for REST API backends.

This connector allows complete interaction with other REST APIs.

Parameters:

**kwargs –

  • host= str – The API host

  • port= int – The API port

  • tls= bool – Whether to use TLS (HTTPS) for API requests

  • validate_cert= bool – Whether to validate TLS certificates (if TLS is True)

  • prefix= str – Path prefix used for all endpoints

  • username= str – Username for basic authentication (optional)

  • password= str – Password for basic authentication (optional)

  • auth_token= str – Bearer token for authentication (optional)

  • restriction= Callable – Restriction filter function (not implemented)

create(o: dict, **kwargs) → str

Create the object by issuing a POST request to the REST API and return the _id

Parameters:

  • o (dict) – The object given (json format)

  • kwargs (dict) – Endpoint options

  • kwargs.endpoint (str | None) – endpoint name relative to self._uri

  • kwargs.method (str | None) – HTTP method option from endpoint model (ignored by create, which always uses POST)

  • kwargs.url_parameters (list | None) – path parameters appended to endpoint

  • kwargs.query_options (dict | None) – query string options appended after ?

  • kwargs.data (dict | list | None) – payload option from endpoint model (ignored by create, which uses o as request payload)

Returns:

the object _id

Return type:

str

Raises:

Error – Raise an error DBError, NotFoundError or any db error

delete_by_id(_id: str, **kwargs) → bool

Delete the object by issuing a DELETE request to the REST API

Parameters:

  • _id (str) – The object _id to delete

  • kwargs (dict) – Endpoint options

  • kwargs.endpoint (str | None) – endpoint name relative to self._uri

  • kwargs.method (str | None) – HTTP method option from endpoint model (ignored by create, which always uses POST)

  • kwargs.url_parameters (list | None) – path parameters appended to endpoint

  • kwargs.query_options (dict | None) – query string options appended after ?

  • kwargs.data (dict | list | None) – payload option from endpoint model (ignored by save, which uses o as request payload)

Returns:

True if the object was successfully deleted

Return type:

bool

Raises:

Error – Raise an error DBError, NotFoundError or any db error

abstractmethod drop(**kwargs)

Drop the collection / table / resource (TO implement in subclasses)

Parameters:

**kwargs –

  • endpoint= str | None – endpoint name relative to self._uri

  • method= str | None – HTTP method option from endpoint model (ignored by create, which always uses POST)

  • url_parameters= list | None – path parameters appended to endpoint

  • query_options= dict | None – query string options appended after ?

  • data= dict | None – payload option from endpoint model (ignored by create,

    which uses o as request payload)

  • username= str – Username for basic authentication (optional)

  • password= str – Password for basic authentication (optional)

  • auth_token= str – Bearer token for authentication (optional)

  • restriction= Callable – Restriction filter function (not implemented)

Returns:

True if the object was successfully deleted

Return type:

bool

Raises:

Error – Raise an error DBError, NotFoundError or any db error

get_by_id(_id: str, **kwargs) → dict

Get the objectby its _id by issuing a GET request to the REST API

Parameters:

  • _id (str) – The object _id to get

  • kwargs (dict) – Endpoint options

  • kwargs.endpoint (str | None) – endpoint name relative to self._uri

  • kwargs.method (str | None) – HTTP method option from endpoint model (ignored by create, which always uses POST)

  • kwargs.url_parameters (list | None) – path parameters appended to endpoint

  • kwargs.query_options (dict | None) – query string options appended after ?

  • kwargs.data (dict | list | None) – payload option from endpoint model (ignored by save, which uses o as request payload)

Returns:

the object corresponding to the _id

Return type:

dict

Raises:

Error – Raise an error DBError, NotFoundError or any db error

save(_id: str, o: dict, **kwargs)

Save / update the object by issuing a PUT request to the REST API and return the _id

Parameters:

  • o (dict) – The object given (json format)

  • kwargs (dict) – Endpoint options

  • kwargs.endpoint (str | None) – endpoint name relative to self._uri

  • kwargs.method (str | None) – HTTP method option from endpoint model (ignored by create, which always uses POST)

  • kwargs.url_parameters (list | None) – path parameters appended to endpoint

  • kwargs.query_options (dict | None) – query string options appended after ?

  • kwargs.data (dict | list | None) – payload option from endpoint model (ignored by save, which uses o as request payload)

Returns:

True if the object was successfully saved/updated

Return type:

bool

Raises:

Error – Raise an error DBError, NotFoundError or any db error

abstractmethod select(select_filter, projection: dict = {}, page_size: int = 0, num_of_element_to_skip: int = 0, sort_object: dict = {}, **kwargs) → list

Select from filter in the DB and return a list of dicts, with pagination (TO implement in subclasses)

Parameters:

  • select_filter – The filter for selection (depends on DB types)

  • projection (dict) – The list of elements we want for each object

  • page_size (int) – number of elements per page

  • num_of_element_to_skip (int) – number of element to skip from beginning

  • sort_object (dict) – Soon

  • kwargs (dict) – Endpoint options

  • kwargs.endpoint (str | None) – endpoint name relative to self._uri

  • kwargs.method (str | None) – HTTP method option from endpoint model (ignored by create, which always uses POST)

  • kwargs.url_parameters (list | None) – path parameters appended to endpoint

  • kwargs.query_options (dict | None) – query string options appended after ?

  • kwargs.data (dict | list | None) – payload option from endpoint model (ignored by save, which uses o as request payload)

Returns:

list of objects matching the selection filter

Return type:

list

Raises:

Error – Raise an error DBError or any db error