How to use predicthq - 10 common examples

def test_location_around_type():
    class SchemaExample(schemas.Model):
        around = schemas.ModelType(schemas.LocationAround)

    m = SchemaExample()

    assert m.import_data({"around": {"origin": '40.730610,-73.935242', "offset": "1km", "scale": "2km", "decay": "0.1"}}).to_primitive() == {'around': {'origin': u'40.730610,-73.935242', 'decay': 0.1, 'scale': u'2km', 'offset': u'1km'}}

    with pytest.raises(schemas.SchematicsDataError):
        m.import_data({"around": "40.730610,-73.935242"}, validate=True)

def test_date_around_type():

    class SchemaExample(schemas.Model):
        around = schemas.ModelType(schemas.DateAround)

    m = SchemaExample()

    assert m.import_data({"around": {"origin": '2020-01-01', "offset": "1d", "scale": "0d", "decay": "0.1"}}).to_primitive() == {'around': {'origin': '2020-01-01', 'decay': 0.1, 'scale': u'0d', 'offset': u'1d'}}

    with pytest.raises(schemas.SchematicsDataError):
        m.import_data({"around": "2020-01-01"}, validate=True)

def test_list_type():

    class SchemaExample(schemas.Model):

        string_list = schemas.ListType(schemas.StringType)

    m = SchemaExample()
    assert m.import_data({"string_list": "string"}).to_primitive() == {"string_list": ["string"]}
    assert m.import_data({"string_list": ["string1", "string2"]}).to_primitive() == {"string_list": ["string1", "string2"]}

def test_datetime_type():

    class SchemaExample(schemas.Model):

        my_datetime = schemas.DateTimeType()

    test_date = datetime(2016, 1, 1, tzinfo=pytz.UTC)
    assert SchemaExample({"my_datetime": "2016-01-01T00:00:00+00:00"}).my_datetime == test_date
    assert SchemaExample({"my_datetime": "2016-01-01T00:00:00+0000"}).my_datetime == test_date
    assert SchemaExample({"my_datetime": "2016-01-01T00:00:00Z"}).my_datetime == test_date
    assert SchemaExample({"my_datetime": test_date}).my_datetime == test_date

def test_string_model_and_string_model_type():

    class MyModel(schemas.StringModel):

        import_format = r"(?P.*)==(?P\d*)"
        export_format = "{left}=={right}"

        left = schemas.StringType()
        right = schemas.IntType()

    class SchemaExample(schemas.Model):

        my_model = schemas.StringModelType(MyModel)

    short_data = {"my_model": "ten==10"}
    long_data = {"my_model": {"left": "ten", "right": 10}}
    model_data = {"my_model": MyModel("ten==10")}
    invalid_data = {"my_model": "10==ten"}

    expected_data = {"my_model": "ten==10"}

    m = SchemaExample()

    assert m.import_data(short_data).to_primitive() == expected_data
    assert m.import_data(long_data).to_primitive() == expected_data

def test_returns(self):

        class SchemaExample(schemas.Model):
            arg1 = schemas.StringType(required=True)
            arg2 = schemas.ListType(schemas.IntType)

        class EndpointExample(BaseEndpoint):

            @decorators.returns(SchemaExample)
            def func(self, **kwargs):
                return kwargs

        endpoint = EndpointExample(None)
        self.assertEqual(endpoint.func(arg1="test", arg2=[1, 2]), SchemaExample({'arg1': 'test', 'arg2': [1, 2]}))

        with self.assertRaises(ValidationError):
            endpoint.func(arg2=[1, 2])

        with self.assertRaises(ValidationError):
            endpoint.func(arg1="value", arg2="invalid")

def wrapper(*args, **kwargs):
            class_name, func_name = re.sub(r'([A-Z]+)', r'_\1', args[0].__class__.__name__).lower().strip('_'), f.__name__
            fixtures = load_reqresp_fixture(req_resp or "{}/{}".format(class_name, func_name))
            with responses.RequestsMock(assert_all_requests_are_fired=False) as rsps:
                for fixture in fixtures:
                    if 'url_re' in fixture:
                        url_re = fixture.pop('url_re')
                        fixture['url'] = re.compile(Client.build_url(url_re))
                    else:
                        fixture['url'] = Client.build_url(fixture['url'])
                    if "content_type" in fixture and fixture['content_type'] == "application/json":
                        fixture['body'] = json.dumps(fixture['body'])
                    rsps.add(**fixture)
                return f(responses=rsps, *args, **kwargs)
        return wrapper

def wrapper(*args, **kwargs):
            return f(client=Client(*client_args, **client_kwargs), *args, **kwargs)
        return wrapper

def setUp(self):
        self.client = predicthq.Client()

def test_construct_with_access_token(self):
        client = predicthq.Client(access_token='token123')
        self.assertEqual(client.access_token, 'token123')