11from datetime import datetime
22from typing import List
33
4+ import requests
5+
46from waste_collection_schedule import Collection
57from waste_collection_schedule .exceptions import (
68 SourceArgumentNotFoundWithSuggestions ,
79)
8- # Include work around for SSL UNSAFE_LEGACY_RENEGOTIATION_DISABLED error
9- from waste_collection_schedule .service .SSLError import get_legacy_session
1010
1111TITLE = "Bath & North East Somerset Council"
1212DESCRIPTION = (
2525 "Garden" : {"icon" : "mdi:leaf" , "alias" : "Garden Waste" },
2626}
2727
28+ API_COLLECTION_SUMMARY_URL = "https://api.bathnes.gov.uk/webapi/api/BinsAPI/v2/BartecFeaturesandSchedules/CollectionSummary/{uprn}"
29+ API_ADDRESSES_SEARCH_URL = "https://api.bathnes.gov.uk/webapi/api/AddressesAPI/v2/search/{postcode}/150/true"
2830
2931class Source :
3032 def __init__ (self , uprn = None , postcode = None , housenameornumber = None ):
@@ -33,18 +35,10 @@ def __init__(self, uprn=None, postcode=None, housenameornumber=None):
3335 self ._uprn = uprn
3436
3537 def fetch (self ) -> List [Collection ]:
36- session = get_legacy_session ()
37-
38- if self ._uprn is None :
39- self ._uprn = self .get_uprn (session )
40-
41- r = session .get (
42- f"https://api.bathnes.gov.uk/webapi/api/BinsAPI/v2/BartecFeaturesandSchedules/CollectionSummary/{ self ._uprn } "
43- )
44- if r .status_code != 200 or r .text .strip () == "" :
45- raise Exception (f"could not get collection summary for uprn { self ._uprn } " )
46- entries = r .json ()
38+ if self ._uprn is None :
39+ self ._uprn = self ._get_uprn ()
4740
41+ entries = self ._call_api (API_COLLECTION_SUMMARY_URL .format (uprn = self ._uprn ))
4842 return [
4943 Collection (
5044 date = datetime .fromisoformat (isodate ).date (),
@@ -57,15 +51,10 @@ def fetch(self) -> List[Collection]:
5751 if (isodate := entry .get (f"{ date_type } CollectionDate" ))
5852 ]
5953
60- def get_uprn (self , session ) -> str :
61- r = session .get (
62- f"https://api.bathnes.gov.uk/webapi/api/AddressesAPI/v2/search/{ self ._postcode } /150/true"
63- )
64- if r .status_code != 200 or r .text .strip () == "" :
65- raise Exception (f"could not get addresses for postcode { self ._postcode } " )
66- addresses = r .json ()
54+ def _get_uprn (self ) -> str :
55+ addresses = self ._call_api (API_ADDRESSES_SEARCH_URL .format (postcode = self ._postcode ))
6756
68- address = next (filter (self .filter_addresses , addresses ), None )
57+ address = next (filter (self ._filter_addresses , addresses ), None )
6958 if address is None :
7059 raise SourceArgumentNotFoundWithSuggestions (
7160 "housenameornumber" ,
@@ -74,5 +63,12 @@ def get_uprn(self, session) -> str:
7463 )
7564 return int (address ["uprn" ])
7665
77- def filter_addresses (self , address ) -> bool :
66+ def _filter_addresses (self , address ) -> bool :
7867 return f"|{ self ._housenameornumber .upper ()} |" in address ["payment_Address" ]
68+
69+ def _call_api (self , url : str ):
70+ r = requests .get (url )
71+ r .raise_for_status ()
72+ if r .text .strip () == "" :
73+ raise Exception (f"empty response" )
74+ return r .json ()
0 commit comments