Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_base_endpoint_init(self):
client = APIClient('username', 'password', 'app_key')
base_endpoint = BaseEndpoint(client)
assert base_endpoint.connect_timeout == 3.05
assert base_endpoint.read_timeout == 16
assert base_endpoint._error == APIError
assert base_endpoint.client == client
def setUp(self):
client = APIClient('username', 'password', 'app_key', 'UK')
self.base_endpoint = BaseEndpoint(client)
response_result_list = {'result': [{}, {}]}
response = self.base_endpoint.process_response(response_result_list, mock_resource, None, False)
assert type(response) == list
assert response[0] == mock_resource()
response_result = {'result': {}}
response = self.base_endpoint.process_response(response_result, mock_resource, None, False)
assert response == mock_resource()
# lightweight tests
response_list = [{}, {}]
response = self.base_endpoint.process_response(response_list, mock_resource, None, True)
assert response == response_list
client = APIClient('username', 'password', 'app_key', lightweight=True)
base_endpoint = BaseEndpoint(client)
response_list = [{}, {}]
response = base_endpoint.process_response(response_list, mock_resource, None, False)
assert type(response) == list
assert response[0] == mock_resource()
import datetime
import requests
from typing import Union
from .baseendpoint import BaseEndpoint
from ..resources import LoginResource
from ..exceptions import LoginError, APIError, InvalidResponse
from ..utils import check_status_code
from ..compat import json_loads
class LoginInteractive(BaseEndpoint):
"""
Interactive Login operations (no certs).
"""
_error = LoginError
def __call__(
self, session: requests.Session = None, lightweight: bool = None
) -> Union[dict, LoginResource]:
"""
Makes login request.
:param requests.session session: Requests session object
:param bool lightweight: If True will return dict not a resource
:rtype: LoginResource
import re
import datetime
import requests
from typing import Union, List
from ..exceptions import APIError, RaceCardError, InvalidResponse
from ..utils import check_status_code
from .baseendpoint import BaseEndpoint
from .. import resources
from ..compat import json_loads
class RaceCard(BaseEndpoint):
"""
RaceCard operations.
"""
app_key = None
def login(self, session: requests.Session = None) -> None:
"""
Parses app key from betfair exchange site.
:param requests.session session: Requests session object
"""
session = session or self.client.session
try:
response = session.get(self.login_url)
except requests.ConnectionError as e:
from .baseendpoint import BaseEndpoint
from .. import resources
from ..utils import clean_locals
from ..filters import time_range
class Account(BaseEndpoint):
"""
Account operations.
"""
URI = 'AccountAPING/v1.0/'
connect_timeout = 6.05
def get_account_funds(self, wallet=None, session=None, lightweight=None):
"""
Get available to bet amount.
:param str wallet: Name of the wallet in question
:param requests.session session: Requests session object
:param bool lightweight: If True will return dict not a resource
:rtype: resources.AccountFunds
from requests import ConnectionError
from ..exceptions import (
APIError,
InvalidResponse,
)
from ..utils import check_status_code
from .baseendpoint import BaseEndpoint
class Navigation(BaseEndpoint):
"""
Navigation operations.
"""
def list_navigation(self, session=None):
"""
This Navigation Data for Applications service allows the retrieval of the
full Betfair market navigation menu from a compressed file.
:param requests.session session: Requests session object
:rtype: json
"""
return self.request(session=session)
def request(self, method=None, params=None, session=None):
import datetime
from requests import ConnectionError
from ..exceptions import (
APIError,
InvalidResponse,
)
from ..utils import check_status_code
from .baseendpoint import BaseEndpoint
from .. import resources
class InPlayService(BaseEndpoint):
"""
In play service operations.
"""
def get_event_timeline(self, event_id, session=None, lightweight=None):
"""
Returns event timeline for event id provided.
:param int event_id: Event id to return
:param requests.session session: Requests session object
:param bool lightweight: If True will return dict not a resource
:rtype: resources.EventTimeline
"""
url = '%s%s' % (self.url, 'eventTimeline')
params = {
from .baseendpoint import BaseEndpoint
from .. import resources
from ..utils import clean_locals
class Scores(BaseEndpoint):
"""
Scores operations.
"""
URI = 'ScoresAPING/v1.0/'
def list_race_details(self, meeting_ids=None, race_ids=None, session=None, lightweight=None):
"""
Search for races to get their details.
:param dict meeting_ids: Optionally restricts the results to the specified meeting IDs.
The unique Id for the meeting equivalent to the eventId for that specific race as
returned by listEvents
:param str race_ids: Optionally restricts the results to the specified race IDs. The
unique Id for the race in the format meetingid.raceTime (hhmm). raceTime is in GMT
:param requests.session session: Requests session object
from .baseendpoint import BaseEndpoint
from .. import resources
from ..filters import (
market_filter,
time_range,
)
from ..utils import clean_locals
class Betting(BaseEndpoint):
"""
Betting operations.
"""
URI = 'SportsAPING/v1.0/'
def list_event_types(self, filter=market_filter(), locale=None, session=None, lightweight=None):
"""
Returns a list of Event Types (i.e. Sports) associated with the markets
selected by the MarketFilter.
:param dict filter: The filter to select desired markets
:param str locale: The language used for the response
:param requests.session session: Requests session object
:param bool lightweight: If True will return dict not a resource