How to use the pycsw.server.Csw function in pycsw

To help you get started, we’ve selected a few pycsw examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github geopython / pycsw / tests / functionaltests / test_suites_functional.py View on Github external
normalize_identifier_fields: bool
        Whether to normalize the identifier fields in responses. This
        parameter is used only in the 'harvesting' and 'manager' suites
    use_xml_canonicalisation: bool
        Whether to compare results with their expected values by using xml
        canonicalisation or simply by doing a diff.
    save_results_directory: str
        Path to a directory where to test results should be saved to. A value
        of ``None`` (the default) means that results will not get saved to
        disk.

    """

    request_environment = _prepare_wsgi_test_environment(request_method,
                                                         request_data)
    pycsw_server = server.Csw(rtconfig=configuration, env=request_environment)
    encoding = "utf-8"
    status, raw_contents = pycsw_server.dispatch_wsgi()
    contents = raw_contents.decode(encoding)
    with codecs.open(expected_result, encoding=encoding) as fh:
        expected = fh.read()
    normalized_result = _normalize(
        contents,
        normalize_identifiers=normalize_identifier_fields
    )
    if use_xml_canonicalisation:
        print("Comparing results using XML canonicalization...")
        matches_expected = _compare_with_xml_canonicalisation(
            normalized_result, expected)
    else:
        print("Comparing results using diffs...")
        matches_expected = _compare_without_xml_canonicalisation(
github OSGeo / grass-addons / grass7 / gui / wxpython / wx.metadata / db.csw.run / db.csw.run.py View on Github external
config = env['PYCSW_CONFIG']

    if env['QUERY_STRING'].lower().find('config') != -1:
        for kvp in env['QUERY_STRING'].split('&'):
            if kvp.lower().find('config') != -1:
                config = kvp.split('=')[1]

    if not os.path.isabs(config):
        config = os.path.join(app_path, config)

    if 'HTTP_HOST' in env and ':' in env['HTTP_HOST']:
        env['HTTP_HOST'] = env['HTTP_HOST'].split(':')[0]

    env['local.app_root'] = app_path

    csw = server.Csw(config, env)

    gzip = False
    if ('HTTP_ACCEPT_ENCODING' in env and
                env['HTTP_ACCEPT_ENCODING'].find('gzip') != -1):
        # set for gzip compressed response
        gzip = True

    # set compression level
    if csw.config.has_option('server', 'gzip_compresslevel'):
        gzip_compresslevel = \
            int(csw.config.get('server', 'gzip_compresslevel'))
    else:
        gzip_compresslevel = 0

    status, contents = csw.dispatch_wsgi()
github planetfederal / registry / registry.py View on Github external
return HttpResponse(message, status=status)

    env = request.META.copy()
    env.update({'local.app_root': os.path.dirname(__file__),
                'REQUEST_URI': request.build_absolute_uri()})

    # pycsw prefers absolute urls, let's get them from the request.
    url = request.build_absolute_uri().split('?')[0]
    PYCSW['server']['url'] = url
    PYCSW['metadata:main']['provider_url'] = url

    # Enable CSW-T when a catalog is defined in the
    if catalog:
        PYCSW['manager']['transactions'] = 'true'

    csw = server.Csw(PYCSW, env)
    csw.orm = 'sqlalchemy'
    status, content = csw.dispatch_wsgi()
    status_code = int(status[0:3])
    response = HttpResponse(content,
                            content_type=csw.contenttype,
                            status=status_code,
                            )

    return response
github geopython / pycsw / pycsw / wsgi.py View on Github external
def application(env, start_response):
    """WSGI wrapper"""

    pycsw_root = get_pycsw_root_path(os.environ, env)
    configuration_path = get_configuration_path(os.environ, env, pycsw_root)
    env['local.app_root'] = pycsw_root
    if 'HTTP_HOST' in env and ':' in env['HTTP_HOST']:
        env['HTTP_HOST'] = env['HTTP_HOST'].split(':')[0]
    csw = server.Csw(configuration_path, env)
    status, contents = csw.dispatch_wsgi()
    headers = {
        'Content-Length': str(len(contents)),
        'Content-Type': str(csw.contenttype)
    }
    if "gzip" in env.get("HTTP_ACCEPT_ENCODING", ""):
        try:
            compression_level = int(
                csw.config.get("server", "gzip_compresslevel"))
            contents, compress_headers = compress_response(
                contents, compression_level)
            headers.update(compress_headers)
        except configparser.NoOptionError:
            print(
                "The client requested a gzip compressed response. However, "
                "the server does not specify the 'gzip_compresslevel' option. "
github azavea / Open-Data-Catalog / OpenDataCatalog / catalog / views.py View on Github external
# update server.url
    server_url = '%s://%s%s' % \
        (request.META['wsgi.url_scheme'],
         request.META['HTTP_HOST'],
         request.META['PATH_INFO'])

    mdict['server']['url'] = server_url

    env = request.META.copy()

    env.update({ 
            'local.app_root': os.path.dirname(__file__),
            'REQUEST_URI': request.build_absolute_uri(),
            })
            
    csw = server.Csw(mdict, env)

    content = csw.dispatch_wsgi()

    return HttpResponse(content, content_type=csw.contenttype)
github cga-harvard / Hypermap-Registry / hypermap / search / views.py View on Github external
# HH should be able to pass env['wsgi.input'] without hanging
    # details at https://github.com/cga-harvard/HHypermap/issues/94
    if request.method == 'POST':
        from StringIO import StringIO
        env['wsgi.input'] = StringIO(request.body)

    env.update({'local.app_root': os.path.dirname(__file__),
                'REQUEST_URI': request.build_absolute_uri()})

    # if this is a catalog based CSW, then update settings
    if url is not None:
        settings.REGISTRY_PYCSW['server']['url'] = url
    if catalog_id is not None:
        settings.REGISTRY_PYCSW['repository']['filter'] = 'catalog_id = %d' % catalog_id

    csw = server.Csw(settings.REGISTRY_PYCSW, env)

    content = csw.dispatch_wsgi()

    # pycsw 2.0 has an API break:
    # pycsw < 2.0: content = xml_response
    # pycsw >= 2.0: content = [http_status_code, content]
    # deal with the API break

    if isinstance(content, list):  # pycsw 2.0+
        content = content[1]

    response = HttpResponse(content, content_type=csw.contenttype)

    # TODO: Fix before 1.0 release. CORS should not be enabled blindly like this.
    response['Access-Control-Allow-Origin'] = '*'
    return response
github GeoNode / geonode / geonode / catalogue / backends / pycsw_local.py View on Github external
mdict = dict(settings.PYCSW['CONFIGURATION'], **CONFIGURATION)
        if 'server' in settings.PYCSW['CONFIGURATION']:
            # override server system defaults with user specified directives
            mdict['server'].update(settings.PYCSW['CONFIGURATION']['server'])
        config = SafeConfigParser()

        for section, options in mdict.iteritems():
            config.add_section(section)
            for option, value in options.iteritems():
                config.set(section, option, value)

        # fake HTTP environment variable
        os.environ['QUERY_STRING'] = ''

        # init pycsw
        csw = server.Csw(config, version='2.0.2')

        # fake HTTP method
        csw.requesttype = 'GET'

        # fake HTTP request parameters
        if identifier is None:  # it's a GetRecords request
            formats = []
            for f in self.catalogue.formats:
                formats.append(METADATA_FORMATS[f][0])

            csw.kvp = {
                'service': 'CSW',
                'version': '2.0.2',
                'elementsetname': 'full',
                'typenames': formats,
                'resulttype': 'results',
github GeoNode / geonode / geonode / catalogue / views.py View on Github external
for group in public_groups:
                if isinstance(group, dict):
                    if 'group' in group:
                        groups_ids.append(group['group'])
                else:
                    groups_ids.append(group.id)

            if len(groups_ids) > 0:
                groups = "(" + (", ".join(str(e) for e in groups_ids)) + ")"
                groups_filter = "(group_id IS NULL OR group_id IN " + groups + ")"
                mdict['repository']['filter'] += " AND " + groups_filter
            else:
                groups_filter = "group_id IS NULL"
                mdict['repository']['filter'] += " AND " + groups_filter

        csw = server.Csw(mdict, env, version='2.0.2')

        content = csw.dispatch_wsgi()

        # pycsw 2.0 has an API break:
        # pycsw < 2.0: content = xml_response
        # pycsw >= 2.0: content = [http_status_code, content]
        # deal with the API break

        if isinstance(content, list):  # pycsw 2.0+
            content = content[1]

        spaces = {'csw': 'http://www.opengis.net/cat/csw/2.0.2',
                  'dc': 'http://purl.org/dc/elements/1.1/',
                  'dct': 'http://purl.org/dc/terms/',
                  'gmd': 'http://www.isotc211.org/2005/gmd',
                  'gml': 'http://www.opengis.net/gml',