Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def make_app(config_file=None, settings=None, loop=None):
from guillotina.asgi import AsgiApp
app = AsgiApp(config_file, settings, loop)
router_klass = app_settings.get("router", TraversalRouter)
app.router = resolve_dotted_name(router_klass)()
# The guillotina application is the last middleware in the chain.
# We instantiate middlewares in reverse order. The last one is the first to be called
last_middleware = app
middlewares = [resolve_dotted_name(m) for m in app_settings.get("middlewares", [])]
for middleware in middlewares[::-1]:
last_middleware = middleware(last_middleware)
return last_middleware
def get_conflict_summary(self, oid, txn, old_serial, writer):
from guillotina.utils import get_current_request
try:
req = get_current_request()
except RequestNotFound:
req = None
max_attempts = app_settings.get("conflict_retry_attempts", 3)
attempts = getattr(req, "_retry_attempt", 0)
return f"""Object ID: {oid}
TID: {txn._tid}
Old Object TID: {old_serial}
Belongs to: {writer.of}
Parent ID: {writer.id}
Retries: {attempts}/{max_attempts}"""
def get_conflict_summary(self, oid, txn, old_serial, writer):
from guillotina.utils import get_current_request
try:
req = get_current_request()
except RequestNotFound:
req = None
max_attempts = app_settings.get("conflict_retry_attempts", 3)
attempts = getattr(req, "_retry_attempt", 0)
return f"""Object ID: {oid}
TID: {txn._tid}
Old Object TID: {old_serial}
Belongs to: {writer.of}
Parent ID: {writer.id}
Retries: {attempts}/{max_attempts}"""
def make_app(config_file=None, settings=None, loop=None):
from guillotina.asgi import AsgiApp
app = AsgiApp(config_file, settings, loop)
router_klass = app_settings.get("router", TraversalRouter)
app.router = resolve_dotted_name(router_klass)()
# The guillotina application is the last middleware in the chain.
# We instantiate middlewares in reverse order. The last one is the first to be called
last_middleware = app
middlewares = [resolve_dotted_name(m) for m in app_settings.get("middlewares", [])]
for middleware in middlewares[::-1]:
last_middleware = middleware(last_middleware)
return last_middleware
def executor(self):
if self._executor is None:
self._executor = ThreadPoolExecutor(max_workers=app_settings.get("thread_pool_workers", 32))
return self._executor
try:
resource, tail = await self.traverse(request)
except (ConflictError, asyncio.CancelledError):
# can also happen from connection errors so we bubble this...
raise
except Exception as _exc:
logger.error("Unhandled exception occurred", exc_info=True)
request.resource = request.tail = None
request.exc = _exc
data = {
"success": False,
"exception_message": str(_exc),
"exception_type": getattr(type(_exc), "__name__", str(type(_exc))), # noqa
}
if app_settings.get("debug"):
data["traceback"] = traceback.format_exc()
raise HTTPBadRequest(content={"reason": data})
request.record("traversed")
request.resource = resource
request.tail = tail
if tail and len(tail) > 0:
# convert match lookups
view_name = routes.path_to_view_name(tail)
elif not tail:
view_name = ""
request.record("beforeauthentication")
authenticated = await authenticate_request(request)
def update_app_settings(settings):
for key, value in settings.items():
if isinstance(app_settings.get(key), dict) and isinstance(value, dict):
app_settings[key].update(value)
else:
app_settings[key] = value
def __init__(self, context, request, field):
self.context = context
self.request = request
self.field = field
iface = resolve_dotted_name(app_settings["cloud_storage"])
alsoProvides(field, iface)
self.file_storage_manager = get_multi_adapter((context, request, field), IFileStorageManager)
self.dm = get_adapter(
self.file_storage_manager, IUploadDataManager, name=app_settings.get("cloud_datamanager") or "db"
)
def load_service(_context, service):
# prevent circular import
from guillotina.security.utils import protect_view
service_conf = service["config"]
factory = resolve_dotted_name(service["klass"])
permission = service_conf.get("permission", app_settings.get("default_permission", None))
protect_view(factory, permission)
method = service_conf.get("method", "GET")
default_layer = resolve_dotted_name(app_settings.get("default_layer", IDefaultLayer))
layer = service_conf.get("layer", default_layer)
name = service_conf.get("name", "")
content = service_conf.get("context", Interface)
logger.debug(
"Defining adapter for " # noqa
"{0:s} {1:s} {2:s} to {3:s} name {4:s}".format(
content.__identifier__,
app_settings["http_methods"][method].__identifier__,
layer.__identifier__,
str(factory),
name,