from collections.abc import Mapping
from abc import ABCMeta
from functools import partial
import importlib
from urllib.parse import urlparse
import requests
import rethinkdb as r
import logging
import logging.config
import os
from sondra import help
from sondra.ref import Reference
from . import signals
DOCSTRING_PROCESSORS = {}
try:
from docutils.core import publish_string
from sphinxcontrib import napoleon
[docs] def google_processor(s):
return publish_string(
str(napoleon.GoogleDocstring(s)),
writer_name='html',
settings_overrides={
"stylesheet_path": "/static/css/flasky.css",
"embed_stylesheet": False,
'report_level': 5
}
)
[docs] def numpy_processor(s):
return publish_string(
str(napoleon.NumpyDocstring(s)),
writer_name='html',
settings_overrides={
"stylesheet_path": "/static/css/flasky.css",
"embed_stylesheet": False,
'report_level': 5
}
)
DOCSTRING_PROCESSORS['google'] = google_processor
DOCSTRING_PROCESSORS['numpy'] = numpy_processor
except ImportError:
pass
try:
from docutils.core import publish_string
DOCSTRING_PROCESSORS['rst'] = partial(publish_string, writer_name='html', settings_overrides={"stylesheet_path": "sondra/css/flasky.css"})
except ImportError:
pass
try:
from markdown import markdown
DOCSTRING_PROCESSORS['markdown'] = markdown
except ImportError:
pass
DOCSTRING_PROCESSORS['preformatted'] = lambda x: "<pre>" + str(x) + "</pre>"
BASIC_TYPES = {
"timedelta": {
"type": "object",
"required": ["start", "end"],
"properties": {
"days": {"type": "integer"},
"hours": {"type": "integer"},
"minutes": {"type": "integer"},
"seconds": {"type": "number"}
}
},
"filterOps": {
"enum": [
'with_fields',
'count',
'max',
'min',
'avg',
'sample',
'sum',
'distinct',
'contains',
'pluck',
'without',
'has_fields',
'order_by',
'between'
]
},
"spatialOps": {
"enum": [
'distance',
'get_intersecting',
'get_nearest',
]
},
"geojsonGeometry": {
"type": "object",
"oneOf": [
{"$ref": "#/definitions/point"},
{"$ref": "#/definitions/lineString"},
{"$ref": "#/definitions/polygon"},
]
},
"geojsonFeature": {
"type": "object",
"properties": {
"type": {"enum": ["Feature"]},
"geometry": {"$ref": "#/definitions/geojsonGeometry"},
"properties": {"type": "object"},
}
},
"geojsonFeatureCollection": {
"type": "object",
"properties": {
"type": {"enum": ["FeatureCollection"]},
"features": {"type": "array", "items": {"$ref": "#/definitions/geojsonFeature"}}
}
},
"pointCoordinates": {
"type": "array",
"items": {"type": "number"},
"minLength": 2,
"maxLength": 3
},
"point": {
"type": "object",
"properties": {
"type": {"enum": ["Point"]},
"coordinates": {"$ref": "#/definitions/pointCoordinates"}
}
},
"lineStringCoordinates": {
"type": "array",
"items": {
"type": "array",
"items": {"$ref": "#/definitions/pointCoordinates"}
},
"minLength": 2
},
"lineString": {
"type": "object",
"properties": {
"type": {"enum": ["Point"]},
"coordinates": {"$ref": "#/definitions/lineStringCoordinates"}
}
},
"polygonCoordinates": {
"type": "array",
"items": {
"type": "array",
"items": {"$ref": "#/definitions/lineStringCoordinates"}
},
"minLength": 4
},
"polygon": {
"type": "object",
"properties": {
"type": {"enum": ["Point"]},
"coordinates": {"$ref": "#/definitions/polygonCoordinates"}
}
}
}
[docs]class SuiteException(Exception):
"""Represents a misconfiguration in a :class:`Suite` class"""
[docs]class Suite(Mapping, metaclass=SuiteMetaclass):
"""This is the "environment" for Sondra. Similar to a `settings.py` file in Django, it defines the
environment in which all :class:`Application`s exist.
The Suite is also a mapping type, and it should be used to access or enumerate all the :class:`Application` objects
that are registered.
Attributes:
always_allowed_formats (set): A set of formats where a
applications (dict): A mapping from application name to Application objects. Suite itself implements a mapping
protocol and this is its backend.
base_url (str): The base URL for the API. The Suite will be mounted off of here.
base_url_scheme (str): http or https, automatically set.
base_url_netloc (str): automatically set hostname of the suite.
connection_config (dict): For each key in connections setup keyword args to be passed to `rethinkdb.connect()`
connections (dict): RethinkDB connections for each key in ``connection_config``
docstring_processor_name (str): Any member of DOCSTRING_PROCESSORS: ``preformatted``, ``rst``, ``markdown``,
``google``, or ``numpy``.
docstring_processor (callable): A ``lambda (str)`` that returns HTML for a docstring.
logging (dict): A dict-config for logging.
log (logging.Logger): A logger object configured with the above dictconfig.
cross_origin (bool=False): Allow cross origin API requests from the browser.
schema (dict): The schema of a suite is a dict where the keys are the names of :class:`Application` objects
registered to the suite. The values are the schemas of the named app. See :class:`Application` for more
details on application schemas.
"""
title = "Sondra-Based API"
name = None
debug = False
applications = None
definitions = BASIC_TYPES
url = "http://localhost:5000/api"
logging = None
docstring_processor_name = 'preformatted'
cross_origin = False
allow_anonymous_formats = {'help', 'schema'}
api_request_processors = ()
connection_config = {
'default': {}
}
working_directory = os.getcwd()
# file system settings
file_upload_permissions = 0o644
file_upload_directory_permissions = 0o755
file_upload_temp_dir = None
file_upload_max_memory_size = 100*2**20
file_upload_handlers = (
'sondra.files.uploadhandler.MemoryFileUploadHandler'
'sondra.files.uploadhandler.TemporaryFileUploadHandler'
)
media_root = 'media'
media_url = '/media'
file_storage = "sondra.files.storage.FileSystemStorage"
@property
[docs] def schema_url(self):
return self.url + ";schema"
@property
[docs] def schema(self):
return {
"id": self.url + ";schema",
"title": self.title,
"type": "object",
"description": self.__doc__ or "*No description provided.*",
"applications": {k: v.url for k, v in self.applications.items()},
"definitions": self.definitions
}
@property
[docs] def full_schema(self):
return {
"id": self.url + ";schema",
"title": self.title,
"type": None,
"description": self.__doc__ or "*No description provided.*",
"applications": {k: v.full_schema for k, v in self.applications.items()},
"definitions": self.definitions
}
def __init__(self):
self.applications = {}
if self.logging:
logging.config.dictConfig(self.logging)
else:
logging.basicConfig(level=logging.DEBUG if self.debug else logging.WARNING)
self.log = logging.getLogger(self.__class__.__name__) # use root logger for the environment
signals.pre_init.send(self.__class__, isntance=self)
self.connections = {name: r.connect(**kwargs) for name, kwargs in self.connection_config.items()}
for name in self.connections:
self.log.warning("Connection established to '{0}'".format(name))
self.log.warning("Suite base url is: '{0}".format(self.url))
self.docstring_processor = DOCSTRING_PROCESSORS[self.docstring_processor_name]
self.log.info('Docstring processor is {0}')
self.name = self.name or self.__class__.__name__
self.description = self.__doc__ or "No description provided."
if isinstance(self.file_storage, str):
mod, cls = self.file_storage.rsplit('.', 1)
mod = importlib.import_module(mod)
cls = getattr(mod, cls)
self.storage = cls(self)
elif isinstance(self.file_storage, type):
self.storage = self.file_storage(self)
else:
self.storage = self.file_storage
signals.post_init.send(self.__class__, instance=self)
[docs] def register_application(self, app):
"""This is called automatically whenever an Application object is constructed."""
if app.slug in self.applications:
self.log.error("Tried to register application '{0}' more than once.".format(app.slug))
raise SuiteException("Tried to register multiple applications with the same name.")
self.applications[app.slug] = app
self.log.info('Registered application {0} to {1}'.format(app.__class__.__name__, app.url))
[docs] def drop_database_objects(self):
for app in self.values():
app.drop_database()
[docs] def ensure_database_objects(self):
for app in self.values():
app.create_database()
app.create_tables()
[docs] def clear_databases(self):
self.drop_database_objects()
self.ensure_database_objects()
[docs] def __getitem__(self, item):
"""Application objects are indexed by "slug." Every Application object registered has its name slugified.
This means that if your app was called `MyCoolApp`, its registered name would be `my-cool-app`. This key is
used whether you are accessing the application via URL or locally via Python. For example, the following
both produce the same result::
URL (yields schema as application/json):
http://localhost:5000/api/my-cool-app;schema
Python (yields schema as a dict):
suite = Suite()
suite['my-cool-app'].schema
"""
return self.applications[item]
def __len__(self):
return len(self.applications)
def __iter__(self):
return iter(self.applications)
def __contains__(self, item):
return item in self.applications
[docs] def help(self, out=None, initial_heading_level=0):
"""Return full reStructuredText help for this class"""
builder = help.SchemaHelpBuilder(self.schema, self.url, out=out, initial_heading_level=initial_heading_level)
builder.begin_subheading(self.name)
builder.define("Suite", self.url)
builder.line()
builder.define("Schema URL", self.schema_url)
builder.line()
builder.build()
builder.line()
builder.begin_subheading("Applications")
builder.begin_list()
for name, coll in self.applications.items():
builder.define(name, coll.url + ';help')
builder.end_list()
builder.end_subheading()
builder.end_subheading()
return builder.rst
[docs] def lookup(self, url):
if not url.startswith(self.url):
return requests.get(url).json() # TODO replace with client.
else:
return Reference(self, url).value
[docs] def lookup_document(self, url):
if not url.startswith(self.url):
return requests.get(url).json() # TODO replace with client.
else:
return Reference(self, url).get_document()