Source code for stardog.admin

"""Administer a Stardog server.
"""

import contextlib2

from . import content_types as content_types
from .http import admin as http_admin


[docs]class Admin(object): """Admin Connection. This is the entry point for admin-related operations on a Stardog server. See Also: https://www.stardog.com/docs/#_administering_stardog """
[docs] def __init__(self, endpoint=None, username=None, password=None): """Initializes an admin connection to a Stardog server. Args: endpoint (str, optional): Url of the server endpoint. Defaults to `http://localhost:5820` username (str, optional): Username to use in the connection. Defaults to `admin` password (str, optional): Password to use in the connection. Defaults to `admin` Examples: >>> admin = Admin(endpoint='http://localhost:9999', username='admin', password='admin') """ self.admin = http_admin.Admin(endpoint, username, password)
[docs] def shutdown(self): """Shuts down the server. """ self.admin.shutdown()
[docs] def database(self, name): """Retrieves an object representing a database. Args: name (str): The database name Returns: Database: The requested database """ return Database(self.admin.database(name))
[docs] def databases(self): """Retrieves all databases. Returns: list[Database]: A list of database objects """ return list(map(Database, self.admin.databases()))
[docs] def new_database(self, name, options=None, *contents): """Creates a new database. Args: name (str): the database name options (dict): Dictionary with database options (optional) *contents (Content or (Content, str), optional): Datasets to perform bulk-load with. Named graphs are made with tuples of Content and the name. Returns: Database: The database object Examples: Options >>> admin.new_database('db', {'search.enabled': True}) bulk-load >>> admin.new_database('db', {}, File('example.ttl'), File('test.rdf')) bulk-load to named graph >>> admin.new_database('db', {}, (File('test.rdf'), 'urn:context')) """ files = [] with contextlib2.ExitStack() as stack: for c in contents: content = c[0] if isinstance(c, tuple) else c context = c[1] if isinstance(c, tuple) else None # we will be opening references to many sources in a # single call use a stack manager to make sure they # all get properly closed at the end data = stack.enter_context(content.data()) files.append({ 'name': content.name, 'content': data, 'content-type': content.content_type, 'content-encoding': content.content_encoding, 'context': context }) return Database(self.admin.new_database(name, options, *files))
[docs] def restore(self, from_path, *, name=None, force=False): """Restore a database. Args: from_path (str): the full path on the server to the backup name (str, optional): the name of the database to restore to if different from the backup force (boolean, optional): a backup will not be restored over an existing database of the same name; the force flag should be used to overwrite the database Examples: >>> admin.restore("/data/stardog/.backup/db/2019-12-01") >>> admin.restore("/data/stardog/.backup/db/2019-11-05", name="db2", force=True) See Also: https://www.stardog.com/docs/#_restore_a_database_from_a_backup """ self.admin.restore(from_path, name=name, force=force)
[docs] def query(self, id): """Gets information about a running query. Args: id (str): Query ID Returns: dict: Query information """ return self.admin.query(id)
[docs] def queries(self): """Gets information about all running queries. Returns: dict: Query information """ return self.admin.queries()
[docs] def kill_query(self, id): """Kills a running query. Args: id (str): ID of the query to kill """ self.admin.kill_query(id)
[docs] def stored_query(self, name): """Retrieves a Stored Query. Args: name (str): The name of the Stored Query to retrieve Returns: StoredQuery: The StoredQuery object """ return StoredQuery(self.admin.stored_query(name))
[docs] def stored_queries(self): """Retrieves all stored queries. Returns: list[StoredQuery]: A list of StoredQuery objects """ return list(map(StoredQuery, self.admin.stored_queries()))
[docs] def new_stored_query(self, name, query, options=None): """Creates a new Stored Query. Args: name (str): The name of the stored query query (str): The query text options (dict, optional): Additional options Returns: StoredQuery: the new StoredQuery Examples: >>> admin.new_stored_query('all triples', 'select * where { ?s ?p ?o . }', { 'database': 'mydb' } ) """ return StoredQuery(self.admin.new_stored_query(name, query, options))
[docs] def clear_stored_queries(self): """Remove all stored queries on the server. """ self.admin.clear_stored_queries()
[docs] def user(self, name): """Retrieves an object representing a user. Args: name (str): The name of the user Returns: User: The User object """ return User(self.admin.user(name))
[docs] def users(self): """Retrieves all users. Returns: list[User]: A list of User objects """ return list(map(User, self.admin.users()))
[docs] def new_user(self, username, password, superuser=False): """Creates a new user. Args: username (str): The username password (str): The password superuser (bool): Should the user be super? Defaults to false. Returns: User: The new User object """ return User(self.admin.new_user(username, password, superuser))
[docs] def role(self, name): """Retrieves an object representing a role. Args: name (str): The name of the Role Returns: Role: The Role object """ return Role(self.admin.role(name))
[docs] def roles(self): """Retrieves all roles. Returns: list[Role]: A list of Role objects """ return list(map(Role, self.admin.roles()))
[docs] def new_role(self, name): """Creates a new role. Args: name (str): The name of the new Role Returns: Role: The new Role object """ return Role(self.admin.new_role(name))
[docs] def virtual_graph(self, name): """Retrieves a Virtual Graph. Args: name (str): The name of the Virtual Graph to retrieve Returns: VirtualGraph: The VirtualGraph object """ return VirtualGraph(self.admin.virtual_graph(name))
[docs] def virtual_graphs(self): """Retrieves all virtual graphs. Returns: list[VirtualGraph]: A list of VirtualGraph objects """ return list(map(VirtualGraph, self.admin.virtual_graphs()))
[docs] def new_virtual_graph(self, name, mappings, options): """Creates a new Virtual Graph. Args: name (str): The name of the virtual graph mappings (Content): New mapping contents options (dict): Options for the new virtual graph Returns: VirtualGraph: the new VirtualGraph Examples: >>> admin.new_virtual_graph( 'users', File('mappings.ttl'), {'jdbc.driver': 'com.mysql.jdbc.Driver'} ) """ with mappings.data() as data: return VirtualGraph( self.admin.new_virtual_graph( name, data.read().decode() if hasattr(data, 'read') else data, options))
[docs] def validate(self): """Validates an admin connection. Returns: bool: The connection state """ return self.admin.validate()
def __enter__(self): return self def __exit__(self, *args): self.admin.__exit__(*args)
[docs]class Database(object): """Database Admin See Also: https://www.stardog.com/docs/#_database_admin """
[docs] def __init__(self, db): """Initializes a Database. Use :meth:`stardog.admin.Admin.database`, :meth:`stardog.admin.Admin.databases`, or :meth:`stardog.admin.Admin.new_database` instead of constructing manually. """ self.db = db
@property def name(self): """The name of the database. """ return self.db.name
[docs] def get_options(self, *options): """Gets database options. Args: *options (str): Database option names Returns: dict: Database options Examples >>> db.get_options('search.enabled', 'spatial.enabled') """ return self.db.get_options(*options)
[docs] def set_options(self, options): """Sets database options. The database must be offline. Args: options (dict): Database options Examples >>> db.set_options({'spatial.enabled': False}) """ self.db.set_options(options)
[docs] def optimize(self): """Optimizes a database. """ self.db.optimize()
[docs] def repair(self): """Repairs a database. The database must be offline. """ self.db.repair()
[docs] def backup(self, *, to=None): """Backup a database. Args: to (string, optional): specify a path on the server to store the backup See Also: https://www.stardog.com/docs/#_backup_a_database """ self.db.backup(to=to)
[docs] def online(self): """Sets a database to online state. """ self.db.online()
[docs] def offline(self): """Sets a database to offline state. """ self.db.offline()
[docs] def copy(self, to): """Makes a copy of this database under another name. The database must be offline. Args: to (str): Name of the new database to be created Returns: Database: The new Database """ db = self.db.copy(to) return Database(db)
[docs] def drop(self): """Drops the database. """ self.db.drop()
[docs]class StoredQuery(object): """Stored Query See Also: https://www.stardog.com/docs/#_list_stored_queries https://www.stardog.com/docs/#_managing_stored_queries """
[docs] def __init__(self, stored_query): """Initializes a stored query. Use :meth:`stardog.admin.Admin.stored_query`, :meth:`stardog.admin.Admin.stored_queries`, or :meth:`stardog.admin.Admin.new_stored_query` instead of constructing manually. """ self.stored_query = stored_query
@property def name(self): """The name of the stored query. """ return self.stored_query.name @property def description(self): """The description of the stored query. """ return self.stored_query.description @property def creator(self): """The creator of the stored query. """ return self.stored_query.creator @property def database(self): """The database the stored query applies to. """ return self.stored_query.database @property def query(self): """The text of the stored query. """ return self.stored_query.query @property def shared(self): """The value of the shared property. """ return self.stored_query.shared @property def reasoning(self): """The value of the reasoning property. """ return self.stored_query.reasoning
[docs] def update(self, **options): """Updates the Stored Query. Args: **options (str): Named arguments to update. Examples: Update description >>> stored_query.update(description='this query finds all the relevant...') """ self.stored_query.update(**options)
[docs] def delete(self): """Deletes the Stored Query. """ self.stored_query.delete()
[docs]class User(object): """User See Also: https://www.stardog.com/docs/#_security """
[docs] def __init__(self, user): """Initializes a User. Use :meth:`stardog.admin.Admin.user`, :meth:`stardog.admin.Admin.users`, or :meth:`stardog.admin.Admin.new_user` instead of constructing manually. """ self.user = user
@property def name(self): """str: The user name. """ return self.user.name
[docs] def set_password(self, password): """Sets a new password. Args: password (str) """ self.user.set_password(password)
[docs] def is_enabled(self): """Checks if the user is enabled. Returns: bool: User activation state """ return self.user.is_enabled()
[docs] def set_enabled(self, enabled): """Enables or disables the user. Args: enabled (bool): Desired User state """ self.user.set_enabled(enabled)
[docs] def is_superuser(self): """Checks if the user is a super user. Returns: bool: Superuser state """ return self.user.is_superuser()
[docs] def roles(self): """Gets all the User's roles. Returns: list[Role] """ return list(map(Role, self.user.roles()))
[docs] def add_role(self, role): """Adds an existing role to the user. Args: role (str or Role): The role to add or its name Examples: >>> user.add_role('reader') >>> user.add_role(admin.role('reader')) """ self.user.add_role(self.__rolename(role))
[docs] def set_roles(self, *roles): """Sets the roles of the user. Args: *roles (str or Role): The roles to add the User to Examples >>> user.set_roles('reader', admin.role('writer')) """ self.user.set_roles(*list(map(self.__rolename, roles)))
[docs] def remove_role(self, role): """Removes a role from the user. Args: role (str or Role): The role to remove or its name Examples >>> user.remove_role('reader') >>> user.remove_role(admin.role('reader')) """ self.user.remove_role(self.__rolename(role))
[docs] def delete(self): """Deletes the user. """ self.user.delete()
[docs] def permissions(self): """Gets the user permissions. See Also: https://www.stardog.com/docs/#_permissions Returns: dict: User permissions """ return self.user.permissions()
[docs] def add_permission(self, action, resource_type, resource): """Add a permission to the user. See Also: https://www.stardog.com/docs/#_permissions Args: action (str): Action type (e.g., 'read', 'write') resource_type (str): Resource type (e.g., 'user', 'db') resource (str): Target resource (e.g., 'username', '*') Examples >>> user.add_permission('read', 'user', 'username') >>> user.add_permission('write', '*', '*') """ self.user.add_permission(action, resource_type, resource)
[docs] def remove_permission(self, action, resource_type, resource): """Removes a permission from the user. See Also: https://www.stardog.com/docs/#_permissions Args: action (str): Action type (e.g., 'read', 'write') resource_type (str): Resource type (e.g., 'user', 'db') resource (str): Target resource (e.g., 'username', '*') Examples >>> user.remove_permission('read', 'user', 'username') >>> user.remove_permission('write', '*', '*') """ self.user.remove_permission(action, resource_type, resource)
[docs] def effective_permissions(self): """Gets the user's effective permissions. Returns: dict: User effective permissions """ return self.user.effective_permissions()
def __rolename(self, role): return role.name if isinstance(role, Role) else role
[docs]class Role(object): """Role See Also: https://www.stardog.com/docs/#_security """
[docs] def __init__(self, role): """Initializes a Role. Use :meth:`stardog.admin.Admin.role`, :meth:`stardog.admin.Admin.roles`, or :meth:`stardog.admin.Admin.new_role` instead of constructing manually. """ self.role = role
@property def name(self): """The name of the Role. """ return self.role.name
[docs] def users(self): """Lists the users for this role. Returns: list[User] """ return list(map(User, self.role.users()))
[docs] def delete(self, force=None): """Deletes the role. Args: force (bool): Force deletion of the role """ self.role.delete(force)
[docs] def permissions(self): """Gets the role permissions. See Also: https://www.stardog.com/docs/#_permissions Returns: dict: Role permissions """ return self.role.permissions()
[docs] def add_permission(self, action, resource_type, resource): """Adds a permission to the role. See Also: https://www.stardog.com/docs/#_permissions Args: action (str): Action type (e.g., 'read', 'write') resource_type (str): Resource type (e.g., 'user', 'db') resource (str): Target resource (e.g., 'username', '*') Examples: >>> role.add_permission('read', 'user', 'username') >>> role.add_permission('write', '*', '*') """ self.role.add_permission(action, resource_type, resource)
[docs] def remove_permission(self, action, resource_type, resource): """Removes a permission from the role. See Also: https://www.stardog.com/docs/#_permissions Args: action (str): Action type (e.g., 'read', 'write') resource_type (str): Resource type (e.g., 'user', 'db') resource (str): Target resource (e.g., 'username', '*') Examples: >>> role.remove_permission('read', 'user', 'username') >>> role.remove_permission('write', '*', '*') """ self.role.remove_permission(action, resource_type, resource)
[docs]class VirtualGraph(object): """Virtual Graph See Also: https://www.stardog.com/docs/#_structured_data """
[docs] def __init__(self, vg): """Initializes a virtual graph. Use :meth:`stardog.admin.Admin.virtual_graph`, :meth:`stardog.admin.Admin.virtual_graphs`, or :meth:`stardog.admin.Admin.new_virtual_graph` instead of constructing manually. """ self.vg = vg
@property def name(self): """The name of the virtual graph. """ return self.vg.name
[docs] def update(self, name, mappings, options): """Updates the Virtual Graph. Args: name (str): The new name mappings (Content): New mapping contents options (dict): New options Examples: >>> vg.update('users', File('mappings.ttl'), {'jdbc.driver': 'com.mysql.jdbc.Driver'}) """ with mappings.data() as data: self.vg.update( name, data.read().decode() if hasattr(data, 'read') else data, options)
[docs] def delete(self): """Deletes the Virtual Graph. """ self.vg.delete()
[docs] def options(self): """Gets Virtual Graph options. Returns: dict: Options """ return self.vg.options()
[docs] def mappings(self, content_type=content_types.TURTLE): """Gets the Virtual Graph mappings. Args: content_type (str): Content type for results. Defaults to 'text/turtle' Returns: str: Mappings in given content type """ return self.vg.mappings(content_type)
[docs] def available(self): """Checks if the Virtual Graph is available. Returns: bool: Availability state """ return self.vg.available()