resource.py 9.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297
  1. import logging
  2. from functools import wraps
  3. from itertools import groupby
  4. from multiprocessing import Process
  5. from threading import Lock, Thread
  6. import arrow
  7. from rdflib import Literal
  8. from rdflib.namespace import XSD
  9. from lakesuperior.config_parser import config
  10. from lakesuperior.exceptions import (
  11. InvalidResourceError, ResourceNotExistsError, TombstoneError)
  12. from lakesuperior.env import env
  13. from lakesuperior.globals import RES_DELETED
  14. from lakesuperior.model.ldp_factory import LDP_NR_TYPE, LdpFactory
  15. from lakesuperior.store.ldp_rs.lmdb_store import TxnManager
  16. logger = logging.getLogger(__name__)
  17. app_globals = env.app_globals
  18. __doc__ = '''
  19. Primary API for resource manipulation.
  20. Quickstart:
  21. >>> # First import default configuration and globals—only done once.
  22. >>> import lakesuperior.default_env
  23. >>> from lakesuperior.api import resource
  24. >>> # Get root resource.
  25. >>> rsrc = resource.get('/')
  26. >>> # Dump graph.
  27. >>> set(rsrc.imr)
  28. {(rdflib.term.URIRef('info:fcres/'),
  29. rdflib.term.URIRef('http://purl.org/dc/terms/title'),
  30. rdflib.term.Literal('Repository Root')),
  31. (rdflib.term.URIRef('info:fcres/'),
  32. rdflib.term.URIRef('http://www.w3.org/1999/02/22-rdf-syntax-ns#type'),
  33. rdflib.term.URIRef('http://fedora.info/definitions/v4/repository#Container')),
  34. (rdflib.term.URIRef('info:fcres/'),
  35. rdflib.term.URIRef('http://www.w3.org/1999/02/22-rdf-syntax-ns#type'),
  36. rdflib.term.URIRef('http://fedora.info/definitions/v4/repository#RepositoryRoot')),
  37. (rdflib.term.URIRef('info:fcres/'),
  38. rdflib.term.URIRef('http://www.w3.org/1999/02/22-rdf-syntax-ns#type'),
  39. rdflib.term.URIRef('http://fedora.info/definitions/v4/repository#Resource')),
  40. (rdflib.term.URIRef('info:fcres/'),
  41. rdflib.term.URIRef('http://www.w3.org/1999/02/22-rdf-syntax-ns#type'),
  42. rdflib.term.URIRef('http://www.w3.org/ns/ldp#BasicContainer')),
  43. (rdflib.term.URIRef('info:fcres/'),
  44. rdflib.term.URIRef('http://www.w3.org/1999/02/22-rdf-syntax-ns#type'),
  45. rdflib.term.URIRef('http://www.w3.org/ns/ldp#Container')),
  46. (rdflib.term.URIRef('info:fcres/'),
  47. rdflib.term.URIRef('http://www.w3.org/1999/02/22-rdf-syntax-ns#type'),
  48. rdflib.term.URIRef('http://www.w3.org/ns/ldp#RDFSource'))}
  49. '''
  50. def transaction(write=False):
  51. '''
  52. Handle atomic operations in a store.
  53. This wrapper ensures that a write operation is performed atomically. It
  54. also takes care of sending a message for each resource changed in the
  55. transaction.
  56. ALL write operations on the LDP-RS and LDP-NR stores go through this
  57. wrapper.
  58. '''
  59. def _transaction_deco(fn):
  60. @wraps(fn)
  61. def _wrapper(*args, **kwargs):
  62. # Mark transaction begin timestamp. This is used for create and
  63. # update timestamps on resources.
  64. env.timestamp = arrow.utcnow()
  65. env.timestamp_term = Literal(env.timestamp, datatype=XSD.dateTime)
  66. with TxnManager(app_globals.rdf_store, write=write) as txn:
  67. ret = fn(*args, **kwargs)
  68. if len(app_globals.changelog):
  69. job = Thread(target=process_queue)
  70. job.start()
  71. logger.debug('Deleting timestamp: {}'.format(getattr(env, 'timestamp')))
  72. delattr(env, 'timestamp')
  73. delattr(env, 'timestamp_term')
  74. return ret
  75. return _wrapper
  76. return _transaction_deco
  77. def process_queue():
  78. '''
  79. Process the message queue on a separate thread.
  80. '''
  81. lock = Lock()
  82. lock.acquire()
  83. while len(app_globals.changelog):
  84. send_event_msg(*app_globals.changelog.popleft())
  85. lock.release()
  86. def send_event_msg(remove_trp, add_trp, metadata):
  87. '''
  88. Break down delta triples, find subjects and send event message.
  89. '''
  90. remove_grp = groupby(remove_trp, lambda x : x[0])
  91. remove_dict = {k[0]: k[1] for k in remove_grp}
  92. add_grp = groupby(add_trp, lambda x : x[0])
  93. add_dict = {k[0]: k[1] for k in add_grp}
  94. subjects = set(remove_dict.keys()) | set(add_dict.keys())
  95. for rsrc_uri in subjects:
  96. logger.info('subject: {}'.format(rsrc_uri))
  97. app_globals.messenger.send
  98. ### API METHODS ###
  99. @transaction()
  100. def get(uid, repr_options={}):
  101. '''
  102. Get an LDPR resource.
  103. The resource comes preloaded with user data and metadata as indicated by
  104. the `repr_options` argument. Any further handling of this resource is done
  105. outside of a transaction.
  106. @param uid (string) Resource UID.
  107. @param repr_options (dict(bool)) Representation options. This is a dict
  108. that is unpacked downstream in the process. The default empty dict results
  109. in default values. The accepted dict keys are:
  110. - incl_inbound: include inbound references. Default: False.
  111. - incl_children: include children URIs. Default: True.
  112. - embed_children: Embed full graph of all child resources. Default: False
  113. '''
  114. rsrc = LdpFactory.from_stored(uid, repr_options)
  115. # Load graph before leaving the transaction.
  116. rsrc.imr
  117. return rsrc
  118. @transaction()
  119. def get_version_info(uid):
  120. '''
  121. Get version metadata (fcr:versions).
  122. '''
  123. return LdpFactory.from_stored(uid).version_info
  124. @transaction()
  125. def get_version(uid, ver_uid):
  126. '''
  127. Get version metadata (fcr:versions).
  128. '''
  129. return LdpFactory.from_stored(uid).get_version(ver_uid)
  130. @transaction(True)
  131. def create(parent, slug, **kwargs):
  132. '''
  133. Mint a new UID and create a resource.
  134. The UID is computed from a given parent UID and a "slug", a proposed path
  135. relative to the parent. The application will attempt to use the suggested
  136. path but it may use a different one if a conflict with an existing resource
  137. arises.
  138. @param parent (string) UID of the parent resource.
  139. @param slug (string) Tentative path relative to the parent UID.
  140. @param **kwargs Other parameters are passed to the
  141. LdpFactory.from_provided method. Please see the documentation for that
  142. method for explanation of individual parameters.
  143. @return string UID of the new resource.
  144. '''
  145. uid = LdpFactory.mint_uid(parent, slug)
  146. logger.debug('Minted UID for new resource: {}'.format(uid))
  147. rsrc = LdpFactory.from_provided(uid, **kwargs)
  148. rsrc.create_or_replace_rsrc(create_only=True)
  149. return uid
  150. @transaction(True)
  151. def create_or_replace(uid, stream=None, **kwargs):
  152. '''
  153. Create or replace a resource with a specified UID.
  154. If the resource already exists, all user-provided properties of the
  155. existing resource are deleted. If the resource exists and the provided
  156. content is empty, an exception is raised (not sure why, but that's how
  157. FCREPO4 handles it).
  158. @param uid (string) UID of the resource to be created or updated.
  159. @param stream (BytesIO) Content stream. If empty, an empty container is
  160. created.
  161. @param **kwargs Other parameters are passed to the
  162. LdpFactory.from_provided method. Please see the documentation for that
  163. method for explanation of individual parameters.
  164. @return string Event type: whether the resource was created or updated.
  165. '''
  166. rsrc = LdpFactory.from_provided(uid, stream=stream, **kwargs)
  167. if not stream and rsrc.is_stored:
  168. raise InvalidResourceError(rsrc.uid,
  169. 'Resource {} already exists and no data set was provided.')
  170. return rsrc.create_or_replace_rsrc()
  171. @transaction(True)
  172. def update(uid, update_str, is_metadata=False):
  173. '''
  174. Update a resource with a SPARQL-Update string.
  175. @param uid (string) Resource UID.
  176. @param update_str (string) SPARQL-Update statements.
  177. @param is_metadata (bool) Whether the resource metadata is being updated.
  178. If False, and the resource being updated is a LDP-NR, an error is raised.
  179. '''
  180. rsrc = LdpFactory.from_stored(uid)
  181. if LDP_NR_TYPE in rsrc.ldp_types:
  182. if is_metadata:
  183. rsrc.patch_metadata(update_str)
  184. else:
  185. raise InvalidResourceError(uid)
  186. else:
  187. rsrc.patch(update_str)
  188. return rsrc
  189. @transaction(True)
  190. def create_version(uid, ver_uid):
  191. '''
  192. Create a resource version.
  193. @param uid (string) Resource UID.
  194. @param ver_uid (string) Version UID to be appended to the resource URI.
  195. NOTE: this is a "slug", i.e. the version URI is not guaranteed to be the
  196. one indicated.
  197. @return string Version UID.
  198. '''
  199. return LdpFactory.from_stored(uid).create_version(ver_uid)
  200. @transaction(True)
  201. def delete(uid, soft=True):
  202. '''
  203. Delete a resource.
  204. @param uid (string) Resource UID.
  205. @param soft (bool) Whether to perform a soft-delete and leave a
  206. tombstone resource, or wipe any memory of the resource.
  207. '''
  208. # If referential integrity is enforced, grab all inbound relationships
  209. # to break them.
  210. refint = app_globals.rdfly.config['referential_integrity']
  211. inbound = True if refint else inbound
  212. repr_opts = {'incl_inbound' : True} if refint else {}
  213. children = app_globals.rdfly.get_descendants(uid)
  214. if soft:
  215. rsrc = LdpFactory.from_stored(uid, repr_opts)
  216. ret = rsrc.bury_rsrc(inbound)
  217. for child_uri in children:
  218. try:
  219. child_rsrc = LdpFactory.from_stored(
  220. app_globals.rdfly.uri_to_uid(child_uri),
  221. repr_opts={'incl_children' : False})
  222. except (TombstoneError, ResourceNotExistsError):
  223. continue
  224. child_rsrc.bury_rsrc(inbound, tstone_pointer=rsrc.uri)
  225. else:
  226. ret = app_globals.rdfly.forget_rsrc(uid, inbound)
  227. for child_uri in children:
  228. child_uid = app_globals.rdfly.uri_to_uid(child_uri)
  229. ret = app_globals.rdfly.forget_rsrc(child_uid, inbound)
  230. return ret
  231. @transaction(True)
  232. def resurrect(uid):
  233. '''
  234. Reinstate a buried (soft-deleted) resource.
  235. @param uid (string) Resource UID.
  236. '''
  237. return LdpFactory.from_stored(uid).resurrect_rsrc()