Refactor ApiClient to use HttpClient (#9114)

* Refactor ApiClient to use HttpClient

* Minor fix

* Refactor and cleanup in ApiConnection

* Add normalized extension methods for string and stream json serialization

* Minor fixes

* Address peer review

* Minor fixes

* Minor fixes and peer review

* Race condition fix

* Minor changes

* Cleanup

* Minor fixes

* Minor fix

Author: jhonabreul

Algorithm.Python/ScheduledQueuingAlgorithm.py

@@ -20,18 +20,18 @@ def initialize(self) -> None:
         self.set_start_date(2020, 9, 1)
         self.set_end_date(2020, 9, 2)
         self.set_cash(100000)
-        
+
         self.__number_of_symbols = 2000
         self.__number_of_symbols_fine = 1000
         self.set_universe_selection(FineFundamentalUniverseSelectionModel(self.coarse_selection_function, self.fine_selection_function, None))
-        
+
         self.set_portfolio_construction(EqualWeightingPortfolioConstructionModel())
-        
+
         self.set_execution(ImmediateExecutionModel())
-        
+
         self._queue = Queue()
         self._dequeue_size = 100
-        
+
         self.add_equity("SPY", Resolution.MINUTE)
         self.schedule.on(self.date_rules.every_day("SPY"), self.time_rules.at(0, 0), self.fill_queue)
         self.schedule.on(self.date_rules.every_day("SPY"), self.time_rules.every(timedelta(minutes=60)), self.take_from_queue)
@@ -40,22 +40,22 @@ def coarse_selection_function(self, coarse: list[CoarseFundamental]) -> list[Sym
         has_fundamentals = [security for security in coarse if security.has_fundamental_data]
         sorted_by_dollar_volume = sorted(has_fundamentals, key=lambda x: x.dollar_volume, reverse=True)
         return [ x.symbol for x in sorted_by_dollar_volume[:self.__number_of_symbols] ]
-    
+
     def fine_selection_function(self, fine: list[FineFundamental]) -> list[Symbol]:
         sorted_by_pe_ratio = sorted(fine, key=lambda x: x.valuation_ratios.pe_ratio, reverse=True)
         return [ x.symbol for x in sorted_by_pe_ratio[:self.__number_of_symbols_fine] ]
-        
+
     def fill_queue(self) -> None:
-        securities = [security for security in self.active_securities.values if security.fundamentals]
-        
+        securities = [security for security in self.active_securities.values() if security.fundamentals]
+
         # Fill queue with symbols sorted by PE ratio (decreasing order)
         self._queue.queue.clear()
         sorted_by_pe_ratio = sorted(securities, key=lambda x: x.fundamentals.valuation_ratios.pe_ratio, reverse=True)
         for security in sorted_by_pe_ratio:
             self._queue.put(security.symbol)
-        
+
     def take_from_queue(self) -> None:
         symbols = [self._queue.get() for _ in range(min(self._dequeue_size, self._queue.qsize()))]
         self.history(symbols, 10, Resolution.DAILY)
-        
+
         self.log(f"Symbols at {self.time}: {[str(symbol) for symbol in symbols]}")

Algorithm/QCAlgorithm.cs

@@ -2484,7 +2484,7 @@ public Option AddOptionContract(Symbol symbol, Resolution? resolution = null, bo
         /// </summary>
         /// <param name="ticker">The currency pair</param>
         /// <param name="resolution">The <see cref="Resolution"/> of market data, Tick, Second, Minute, Hour, or Daily. Default is <see cref="Resolution.Minute"/></param>
-        /// <param name="market">The foreign exchange trading market, <seealso cref="Market"/>. Default value is null and looked up using <see cref="IBrokerageModel.DefaultMarkets"> in <see cref="AddSecurity{T}"/></param>
+        /// <param name="market">The foreign exchange trading market, <seealso cref="Market"/>. Default value is null and looked up using <see cref="IBrokerageModel.DefaultMarkets" /> in <see cref="AddSecurity{T}"/></param>
         /// <param name="fillForward">If true, returns the last available data even if none in that timeslice. Default is <value>true</value></param>
         /// <param name="leverage">The requested leverage for this forex security. Default is set by <see cref="SecurityInitializer"/></param>
         /// <returns>The new <see cref="Forex"/> security</returns>
@@ -2499,7 +2499,7 @@ public Forex AddForex(string ticker, Resolution? resolution = null, string marke
         /// </summary>
         /// <param name="ticker">The CFD ticker symbol</param>
         /// <param name="resolution">The <see cref="Resolution"/> of market data, Tick, Second, Minute, Hour, or Daily. Default is <see cref="Resolution.Minute"/></param>
-        /// <param name="market">The cfd trading market, <seealso cref="Market"/>. Default value is null and looked up using <see cref="IBrokerageModel.DefaultMarkets"> in <see cref="AddSecurity{T}"/></param>
+        /// <param name="market">The cfd trading market, <seealso cref="Market"/>. Default value is null and looked up using <see cref="IBrokerageModel.DefaultMarkets"/> in <see cref="AddSecurity{T}"/></param>
         /// <param name="fillForward">If true, returns the last available data even if none in that timeslice. Default is <value>true</value></param>
         /// <param name="leverage">The requested leverage for this CFD. Default is set by <see cref="SecurityInitializer"/></param>
         /// <returns>The new <see cref="Cfd"/> security</returns>
@@ -2530,7 +2530,7 @@ public Index AddIndex(string ticker, Resolution? resolution = null, string marke
         /// </summary>
         /// <param name="ticker">The crypto ticker symbol/param>
         /// <param name="resolution">The <see cref="Resolution"/> of market data, Tick, Second, Minute, Hour, or Daily. Default is <see cref="Resolution.Minute"/></param>
-        /// <param name="market">The The crypto trading market, <seealso cref="Market"/>. Default value is null and looked up using <see cref="IBrokerageModel.DefaultMarkets"> in <see cref="AddSecurity{T}"/></param>
+        /// <param name="market">The The crypto trading market, <seealso cref="Market"/>. Default value is null and looked up using <see cref="IBrokerageModel.DefaultMarkets"/> in <see cref="AddSecurity{T}"/></param>
         /// <param name="fillForward">If true, returns the last available data even if none in that timeslice. Default is <value>true</value></param>
         /// <param name="leverage">The requested leverage for this crypto. Default is set by <see cref="SecurityInitializer"/></param>
         /// <returns>The new <see cref="Crypto"/> security</returns>
@@ -2545,7 +2545,7 @@ public Crypto AddCrypto(string ticker, Resolution? resolution = null, string mar
         /// </summary>
         /// <param name="ticker">The crypto future ticker symbol</param>
         /// <param name="resolution">The <see cref="Resolution"/> of market data, Tick, Second, Minute, Hour, or Daily. Default is <see cref="Resolution.Minute"/></param>
-        /// <param name="market">The The crypto future trading market, <seealso cref="Market"/>. Default value is null and looked up using <see cref="IBrokerageModel.DefaultMarkets"> in <see cref="AddSecurity{T}"/></param>
+        /// <param name="market">The The crypto future trading market, <seealso cref="Market"/>. Default value is null and looked up using <see cref="IBrokerageModel.DefaultMarkets"/> in <see cref="AddSecurity{T}"/></param>
         /// <param name="fillForward">If true, returns the last available data even if none in that timeslice. Default is <value>true</value></param>
         /// <param name="leverage">The requested leverage for this crypto future. Default is set by <see cref="SecurityInitializer"/></param>
         /// <returns>The new <see cref="CryptoFuture"/> security</returns>

Api/ApiConnection.cs

@@ -15,11 +15,15 @@
 
 using System;
 using RestSharp;
-using Newtonsoft.Json;
-using QuantConnect.Orders;
 using QuantConnect.Logging;
 using System.Threading.Tasks;
-using RestSharp.Authenticators;
+using System.Net.Http;
+using System.Net.Http.Headers;
+using System.Text;
+using System.Collections.Generic;
+using QuantConnect.Util;
+using System.IO;
+using System.Threading;
 
 namespace QuantConnect.Api
 {
@@ -28,11 +32,15 @@ namespace QuantConnect.Api
     /// </summary>
     public class ApiConnection
     {
-        private readonly static JsonSerializerSettings _jsonSettings = new() { Converters = { new LiveAlgorithmResultsJsonConverter(), new OrderJsonConverter() } };
+        /// <summary>
+        /// Authorized client to use for requests.
+        /// </summary>
+        private HttpClient _httpClient;
 
         /// <summary>
         /// Authorized client to use for requests.
         /// </summary>
+        [Obsolete("RestSharp is deprecated and will be removed in a future release. Please use the SetClient method or the request methods that take an HttpRequestMessage")]
         public RestClient Client { get; set; }
 
         // Authorization Credentials
@@ -46,11 +54,14 @@ public class ApiConnection
         /// </summary>
         /// <param name="userId">User Id number from QuantConnect.com account. Found at www.quantconnect.com/account </param>
         /// <param name="token">Access token for the QuantConnect account. Found at www.quantconnect.com/account </param>
-        public ApiConnection(int userId, string token)
+        /// <param name="baseUrl">The client's base address</param>
+        /// <param name="defaultHeaders">Default headers for the client</param>
+        /// <param name="timeout">The client timeout in seconds</param>
+        public ApiConnection(int userId, string token, string baseUrl = null, Dictionary<string, string> defaultHeaders = null, int timeout = 0)
         {
             _token = token;
             _userId = userId.ToStringInvariant();
-            Client = new RestClient(Globals.Api);
+            SetClient(!string.IsNullOrEmpty(baseUrl) ? baseUrl : Globals.Api, defaultHeaders, timeout);
         }
 
         /// <summary>
@@ -60,13 +71,40 @@ public bool Connected
         {
             get
             {
-                var request = new RestRequest("authenticate", Method.GET);
-                AuthenticationResponse response;
-                if (TryRequest(request, out response))
+                using var request = new HttpRequestMessage(HttpMethod.Get, "authenticate");
+                return TryRequest(request, out AuthenticationResponse response) && response.Success;
+            }
+        }
+
+        /// <summary>
+        /// Overrides the current client
+        /// </summary>
+        /// <param name="baseUrl">The client's base address</param>
+        /// <param name="defaultHeaders">Default headers for the client</param>
+        /// <param name="timeout">The client timeout in seconds</param>
+        public void SetClient(string baseUrl, Dictionary<string, string> defaultHeaders = null, int timeout = 0)
+        {
+            if (_httpClient != null)
+            {
+                _httpClient.DisposeSafely();
+            }
+
+            _httpClient = new HttpClient() { BaseAddress = new Uri($"{baseUrl.TrimEnd('/')}/") };
+            Client = new RestClient(baseUrl);
+
+            if (defaultHeaders != null)
+            {
+                foreach (var header in defaultHeaders)
                 {
-                    return response.Success;
+                    _httpClient.DefaultRequestHeaders.Add(header.Key, header.Value);
                 }
-                return false;
+                Client.AddDefaultHeaders(defaultHeaders);
+            }
+
+            if (timeout > 0)
+            {
+                _httpClient.Timeout = TimeSpan.FromSeconds(timeout);
+                Client.Timeout = timeout * 1000;
             }
         }
 
@@ -77,6 +115,7 @@ public bool Connected
         /// <param name="request"></param>
         /// <param name="result">Result object from the </param>
         /// <returns>T typed object response</returns>
+        [Obsolete("RestSharp is deprecated and will be removed in a future release. Please use the TryRequest(HttpRequestMessage)")]
         public bool TryRequest<T>(RestRequest request, out T result)
             where T : RestResponse
         {
@@ -90,7 +129,24 @@ public bool TryRequest<T>(RestRequest request, out T result)
         /// </summary>
         /// <typeparam name="T"></typeparam>
         /// <param name="request"></param>
+        /// <param name="result">Result object from the </param>
+        /// <param name="timeout">Timeout for the request</param>
         /// <returns>T typed object response</returns>
+        public bool TryRequest<T>(HttpRequestMessage request, out T result, TimeSpan? timeout = null)
+            where T : RestResponse
+        {
+            var resultTuple = TryRequestAsync<T>(request).SynchronouslyAwaitTaskResult();
+            result = resultTuple.Item2;
+            return resultTuple.Item1;
+        }
+
+        /// <summary>
+        /// Place a secure request and get back an object of type T.
+        /// </summary>
+        /// <typeparam name="T"></typeparam>
+        /// <param name="request"></param>
+        /// <returns>T typed object response</returns>
+        [Obsolete("RestSharp is deprecated and will be removed in a future release. Please use the TryRequestAsync(HttpRequestMessage)")]
         public async Task<Tuple<bool, T>> TryRequestAsync<T>(RestRequest request)
             where T : RestResponse
         {
@@ -116,7 +172,7 @@ public async Task<Tuple<bool, T>> TryRequestAsync<T>(RestRequest request)
                 }
 
                 responseContent = restsharpResponse.Content;
-                result = JsonConvert.DeserializeObject<T>(responseContent, _jsonSettings);
+                result = responseContent.DeserializeJson<T>();
 
                 if (result == null || !result.Success)
                 {
@@ -133,36 +189,133 @@ public async Task<Tuple<bool, T>> TryRequestAsync<T>(RestRequest request)
             return new Tuple<bool, T>(true, result);
         }
 
+        /// <summary>
+        /// Place a secure request and get back an object of type T.
+        /// </summary>
+        /// <typeparam name="T"></typeparam>
+        /// <param name="request"></param>
+        /// <param name="timeout">Timeout for the request</param>
+        /// <returns>T typed object response</returns>
+        public async Task<Tuple<bool, T>> TryRequestAsync<T>(HttpRequestMessage request, TimeSpan? timeout = null)
+            where T : RestResponse
+        {
+            HttpResponseMessage response = null;
+            Stream responseContentStream = null;
+            T result = null;
+            try
+            {
+                if (request.RequestUri.OriginalString.StartsWith('/'))
+                {
+                    request.RequestUri = new Uri(request.RequestUri.ToString().TrimStart('/'), UriKind.Relative);
+                }
+
+                SetAuthenticator(request);
+
+                // Execute the authenticated REST API Call
+                if (timeout.HasValue)
+                {
+                    using var cancellationTokenSource = new CancellationTokenSource(timeout.Value);
+                    response = await _httpClient.SendAsync(request, cancellationTokenSource.Token).ConfigureAwait(false);
+                    responseContentStream = await response.Content.ReadAsStreamAsync(cancellationTokenSource.Token).ConfigureAwait(false);
+                }
+                else
+                {
+                    response = await _httpClient.SendAsync(request).ConfigureAwait(false);
+                    responseContentStream = await response.Content.ReadAsStreamAsync().ConfigureAwait(false);
+                }
+
+                result = responseContentStream.DeserializeJson<T>(leaveOpen: true);
+
+                if (!response.IsSuccessStatusCode)
+                {
+                    Log.Error($"ApiConnect.TryRequest({request.RequestUri}): HTTP Error: {(int)response.StatusCode} {response.ReasonPhrase}. " +
+                        $"Content: {GetRawResponseContent(responseContentStream)}");
+                }
+                if (result == null || !result.Success)
+                {
+                    if (Log.DebuggingEnabled)
+                    {
+                        Log.Debug($"ApiConnection.TryRequest({request.RequestUri}): Raw response: '{GetRawResponseContent(responseContentStream)}'");
+                    }
+                    return new Tuple<bool, T>(false, result);
+                }
+            }
+            catch (Exception err)
+            {
+                Log.Error($"ApiConnection.TryRequest({request.RequestUri}): Error: {err.Message}, Response content: {GetRawResponseContent(responseContentStream)}");
+                return new Tuple<bool, T>(false, null);
+            }
+            finally
+            {
+                response?.DisposeSafely();
+                responseContentStream?.DisposeSafely();
+            }
+
+            return new Tuple<bool, T>(true, result);
+        }
+
+        private static string GetRawResponseContent(Stream stream)
+        {
+            if (stream == null)
+            {
+                return string.Empty;
+            }
+
+            try
+            {
+                stream.Position = 0;
+                using var reader = new StreamReader(stream, leaveOpen: true);
+                return reader.ReadToEnd();
+            }
+            catch (Exception)
+            {
+                return string.Empty;
+            }
+        }
+
         private void SetAuthenticator(RestRequest request)
         {
-            var newTimeStamp = (int)Time.TimeStamp();
+            var base64EncodedAuthenticationString = GetAuthenticatorHeader(out var timeStamp);
+            request.AddHeader("Authorization", $"Basic {base64EncodedAuthenticationString}");
+            request.AddHeader("Timestamp", timeStamp);
+        }
 
+        private void SetAuthenticator(HttpRequestMessage request)
+        {
+            var base64EncodedAuthenticationString = GetAuthenticatorHeader(out var timeStamp);
+            _httpClient.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Basic", base64EncodedAuthenticationString);
+            request.Headers.Add("Timestamp", timeStamp);
+        }
+
+        private string GetAuthenticatorHeader(out string timeStamp)
+        {
+            var newTimeStamp = (int)Time.TimeStamp();
             var currentAuth = _authenticator;
             if (currentAuth == null || newTimeStamp - currentAuth.TimeStamp > 7000)
             {
                 // Generate the hash each request
                 // Add the UTC timestamp to the request header.
                 // Timestamps older than 7200 seconds will not work.
                 var hash = Api.CreateSecureHash(newTimeStamp, _token);
-                var authenticator = new HttpBasicAuthenticator(_userId, hash);
-                _authenticator = currentAuth = new LeanAuthenticator(authenticator, newTimeStamp);
-
-                Client.Authenticator = currentAuth.Authenticator;
+                var authenticationString = $"{_userId}:{hash}";
+                var base64EncodedAuthenticationString = Convert.ToBase64String(Encoding.UTF8.GetBytes(authenticationString));
+                _authenticator = currentAuth = new LeanAuthenticator(newTimeStamp, base64EncodedAuthenticationString);
             }
 
-            request.AddHeader("Timestamp", currentAuth.TimeStampStr);
+            timeStamp = currentAuth.TimeStampStr;
+            return currentAuth.Base64EncodedAuthenticationString;
         }
 
         private class LeanAuthenticator
         {
             public int TimeStamp { get; }
             public string TimeStampStr { get; }
-            public HttpBasicAuthenticator Authenticator { get; }
-            public LeanAuthenticator(HttpBasicAuthenticator authenticator, int timeStamp)
+            public string Base64EncodedAuthenticationString { get; }
+            public LeanAuthenticator(int timeStamp, string base64EncodedAuthenticationString)
             {
                 TimeStamp = timeStamp;
-                Authenticator = authenticator;
                 TimeStampStr = timeStamp.ToStringInvariant();
+                Base64EncodedAuthenticationString = base64EncodedAuthenticationString;
             }
         }
     }