Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@rpc(Unicode, Integer, _returns=Iterable(Unicode))
def ask_question(ctx, question, answer):
"""Ask Stan a question!
<b>Ask Stan questions as a Service</b>
@param name the name to say hello to
@param times the number of times to say hello
@return the completed array
"""
yield u'To an artificial mind, all reality is virtual. How do they know that the real world isn\'t just another simulation? How do you?'
@rpc(Unicode(values=['A', 'B', 'C']), _returns=A)
def get_some_a(self, type_name):
if type_name == 'A':
return A(i=1)
if type_name == 'B':
return B(i=2, s='s')
if type_name == 'C':
return C(i=3, d=datetime.utcnow())
def get_xml_as_object(*_, **__):
raise _local_import_error
from sqlalchemy.sql.type_api import UserDefinedType
from spyne import ComplexModel, ValidationError, Unicode
from spyne.util import six
from spyne.util.six import binary_type, text_type
from spyne.util.fileproxy import SeekableFileProxy
class FileData(ComplexModel):
_type_info = [
('name', Unicode),
('type', Unicode),
('path', Unicode),
]
class PGXml(UserDefinedType):
def __init__(self, pretty_print=False, xml_declaration=False,
encoding='UTF-8'):
super(PGXml, self).__init__()
self.xml_declaration = xml_declaration
self.pretty_print = pretty_print
self.encoding = encoding
def get_col_spec(self, **_):
return "xml"
def bind_processor(self, dialect):
__table_args__ = {"sqlite_autoincrement": True}
id = UnsignedInteger32(primary_key=True)
application = Unicode(256)
operation = Unicode(256)
class User(TableModel):
__tablename__ = 'user'
__namespace__ = 'spyne.examples.sql_crud'
__table_args__ = {"sqlite_autoincrement": True}
id = UnsignedInteger32(primary_key=True)
name = Unicode(256)
first_name = Unicode(256)
last_name = Unicode(256)
permissions = Array(Permission, store_as='table')
@memoize
def TCrudService(T, T_name):
class CrudService(Service):
@rpc(Mandatory.UnsignedInteger32, _returns=T)
def get(ctx, obj_id):
return ctx.udc.session.query(T).filter_by(id=obj_id).one()
@rpc(T, _returns=UnsignedInteger32)
def put(ctx, obj):
if obj.id is None:
ctx.udc.session.add(obj)
ctx.udc.session.flush()
@rpc(Unicode, Integer, _returns=Iterable(Unicode))
def say_hello(ctx, name, times):
for i in range(times):
yield u'Hello, %s' % name
# HACK!: sometimes non-db attributes restored from database don't
# get properly reinitialized.
if isclass(v) and issubclass(v, ModelBase):
continue
polymap = t.Attributes.polymap
if polymap is not None:
t = polymap.get(v.__class__, t)
if v is not None:
retval.append("%s=%s" % (k, log_repr(v, t, parent=k, tags=tags)))
i += 1
return "%s(%s)" % (cls.get_type_name(), ', '.join(retval))
if issubclass(cls, Unicode) and isinstance(obj, six.string_types):
if len(obj) > MAX_STRING_FIELD_LENGTH:
return '%r(...)' % obj[:MAX_STRING_FIELD_LENGTH]
return repr(obj)
if issubclass(cls, File) and isinstance(obj, FileData):
return log_repr(obj, FileData, tags=tags)
retval = repr(obj)
if len(retval) > MAX_STRING_FIELD_LENGTH:
retval = retval[:MAX_STRING_FIELD_LENGTH] + "(...)"
return retval
@rpc(Unicode, Integer, _returns=Iterable(Unicode))
def say_hello_as_file(ctx, name, times):
if isinstance(ctx.transport, HttpTransportContext):
file_name = "{}.xml".format(ctx.descriptor.name)
ctx.transport.set_mime_type("application/xml")
# Force download
ctx.transport.add_header('Content-Disposition', 'attachment',
filename=file_name)
for i in range(times):
yield u'Hello, %s' % name
__namespace__ = 'spyne.examples.sql_crud'
__table_args__ = {"sqlite_autoincrement": True}
id = UnsignedInteger32(primary_key=True)
application = Unicode(256)
operation = Unicode(256)
class User(TableModel):
__tablename__ = 'user'
__namespace__ = 'spyne.examples.sql_crud'
__table_args__ = {"sqlite_autoincrement": True}
id = UnsignedInteger32(primary_key=True)
name = Unicode(256)
first_name = Unicode(256)
last_name = Unicode(256)
permissions = Array(Permission, store_as='table')
@memoize
def TCrudService(T, T_name):
class CrudService(Service):
@rpc(Mandatory.UnsignedInteger32, _returns=T)
def get(ctx, obj_id):
return ctx.udc.session.query(T).filter_by(id=obj_id).one()
@rpc(T, _returns=UnsignedInteger32)
def put(ctx, obj):
if obj.id is None:
ctx.udc.session.add(obj)
ctx.udc.session.flush()
class Property(ComplexModel):
"""Property description (used by spyne)."""
__namespace__ = "urn:schemas-microsoft-com:xml-analysis"
_type_info = {
"LocaleIdentifier": Unicode,
"Format": Unicode,
"Catalog": Unicode,
"Content": Unicode,
"DataSourceInfo": Unicode,
"Password": Unicode,
"StateSupport": Unicode,
"Timeout": Unicode,
"ProviderVersion": Unicode,
"BASE_CUBE_NAME": Unicode,
"AxisFormat": Unicode,
"BeginRange": Unicode,
"EndRange": Unicode,
"MDXSupport": Unicode,
"ProviderName": Unicode,
"UserName": Unicode,
}
class Restriction(ComplexModel):
"""Restriction description (used by spyne)."""
__namespace__ = "urn:schemas-microsoft-com:xml-analysis"
_type_info = {
"CATALOG_NAME": Unicode,