eshangj/stackoverflow_q_and_a_sample huggingface dataset:
Question: I am very fond of the returns library in python, and I would like to use it more. I have a little issue right now. Currently, I have a function that uses a redis client and gets the value corresponding to a key, like so: from redis import Redis from returns.context import RequiresContext def get_value(key: str) -> RequiresContext[str, Redis]: def inner(client: Redis) -> str: value = client.get(key) return value.decode("utf-8") return RequiresContext(inner)  Obviously, that function works like a charm: with Redis( host=redis_host, port=redis_port, password=redis_password, ) as redis_client: value = get_value(key="my-key")(redis_client) print("value = ", value)  Now, I would like to use the asyncio pendant of that code, i.e. use the redis.asyncio.Redis. Unfortunately, it looks like things become a bit more complicated in that case. I should probably switch from RequiresContext to RequiresContextFutureResultE, but I was not able to find a working solution. Here's the best code I was able to come up with: async def get_value(key: str) -> RequiresContextFutureResultE[str, Redis]: async def inner(client: Redis) -> FutureResultE[str]: value = await client.get(key) return FutureResult.from_value(value.decode("utf-8")) return RequiresContextFutureResultE(inner)  When I run it like this: async def main(): async with Redis( host="localhost", port=6379, password="902nks291", db=15, ) as redis_client: rcfr = get_value(key="user-id") value = await rcfr(redis_client) print("value: ", value) asyncio.run(main())  I get the error that rcfr is not a callable. Can someone help me figure out how I should fix my code to make it work the way I want?

Answer: If you call a function (get_value) defined with async def you get an awaitable which you must use with await to get its return value. That's why you get the error. But get_value shouldn't be async def. It just defines and returns a function (wrapped by RequiresContextFutureResultE), it doesn't perform any IO itself.
Question: I want to use Welford's method to compute a running variance and mean. I came across this implementation of Welford's method in Python. However, when testing to double-check that it results in the same output as the standard Numpy implementation of calculating variance, I do find that there is a difference in output. Running the following code (using the python module unittest) shows that the two give different results (even after testing many times): random_sample = np.random.normal(0, 1, 100) std = np.var(random_sample, dtype=np.longdouble) mean = np.mean(random_sample, dtype=np.longdouble) welford = Welford() welford.add_all(random_sample) self.assertAlmostEqual(mean, welford.mean) self.assertAlmostEqual(var, welford.var_s) >> AssertionError: 1.1782075496578717837 != 1.1901086360180526 within 7 places (0.011901086360180828804 difference)  Interestingly, there is only a difference in the variance, not the mean. For my purposes, a 0.012 difference is significant enough that it could affect my results. Why would there be such a difference? Could this be due to compounding floating point errors? If so, would my best bet be to rewrite the package to use the Decimal class?

Answer: By default, np.var computes the so-called "population variance", in which the number of degrees of freedom is equal to the number of elements in the array. wellford.var_s is the sample variance, in which the number of degrees of freedom is the number of elements in the array minus one. To eliminate the discrepancy, pass ddof=1 to np.var: import numpy as np from welford import Welford random_sample = np.random.normal(0, 1, 100) var = np.var(random_sample, dtype=np.longdouble, ddof=1) welford = Welford() welford.add_all(random_sample) np.testing.assert_allclose(var, welford.var_s)  Alternatively, if it is appropriate to use the population variance in your application, use welford.var_p. var = np.var(random_sample, dtype=np.longdouble) np.testing.assert_allclose(var, welford.var_p)  For a description of the difference between the two, see the development version of the np.var documentation.
Question: I created an agent with No Code tools offered by the Agent Builder GUI: https://vertexaiconversation.cloud.google.com/ I created a playbook and added a few Data store Tools for the agent to use for RAG. I'd like to call this agent programmatically to integrate it into mobile apps or web pages. There's a lot of code related to the classic Dialogflow agents, the Agent Builder is quite new and uses the Gemini 1.0 Pro under the hood. I've seen this code https://stackoverflow.com/a/78229704/292502 however the question was about DialogFlow ES while the Agent Builder agent is rather a DialogFLow CX agent under the hood (and is listed in the Dialogflow CX dashboard). The Python package is promising, but I haven't found how can I have a conversation with the agent Playbook after I get hold of one. Or maybe I'm just looking at the wrong place. I was also browsing https://github.com/GoogleCloudPlatform/python-docs-samples/tree/main/dialogflow-cx but webhooks, intents and fulfillments are for the "classic" agents. I tried to go over https://github.com/googleapis/google-cloud-python/blob/main/packages/google-cloud-dialogflow-cx/samples/generated_samples/ but haven't find the one which would help me yet.

Answer: Looks like the "traditional" Dialogflow CX API calls can invoke an Agent Builder agent: import uuid from google.cloud.dialogflowcx_v3beta1.services.agents import AgentsClient from google.cloud.dialogflowcx_v3beta1.services.sessions import SessionsClient from google.cloud.dialogflowcx_v3beta1.types import session PROJECT_ID = "your-project-name" LOCATION_ID = "project-region" # example: us-central1 AGENT_ID = "a uuid of the agent" AGENT = f"projects/{PROJECT_ID}/locations/{LOCATION_ID}/agents/{AGENT_ID}" LANGUAGE_CODE = "en-us" SESSION_ID = uuid.uuid4() session_path = f"{AGENT}/sessions/{SESSION_ID}" print(f"Session path: {session_path}\n") client_options = None agent_components = AgentsClient.parse_agent_path(AGENT) location_id = agent_components["location"] if location_id != "global": api_endpoint = f"{LOCATION_ID}-dialogflow.googleapis.com:443" print(f"API Endpoint: {api_endpoint}\n") client_options = {"api_endpoint": api_endpoint} session_client = SessionsClient(client_options=client_options) text = "Your test prompt" text_input = session.TextInput(text=text) query_input = session.QueryInput(text=text_input, language_code=LANGUAGE_CODE) request = session.DetectIntentRequest( session=session_path, query_input=query_input ) response = session_client.detect_intent(request=request) print("=" * 20) print(f"Query text: {response.query_result.text}") response_messages = [ " ".join(msg.text.text) for msg in response.query_result.response_messages ] print(f"Response text: {' '.join(response_messages)}\n")
Question: I am unit testing a python code. It has an asyncio loop. The main two functions in this loop have endless while loops. I have tested the non-async parts of the code. I am wondering what is the best strategy to unit test these two functions: import asyncio def main(): loop = asyncio.get_event_loop() loop.run_until_complete(run_async_tasks()) . . async def run_async_tasks(): # initialize loop and its variables loop = asyncio.get_event_loop() queue = asyncio.Queue(maxsize=100) # Creating Asyncio tasks task1 = loop.create_task(receive_data(queue), name="receive_data") task2 = loop.create_task(analyse_data(queue), name="analyse_data") await asyncio.gather(task1, task2) async def receive_data(queue): # Data packets are yielded from an non-ending stream by sync_receive_data_packets for data_packet in sync_receive_data_packets(): await queue.put(data_packet) async def analyse_data(queue): while not termination_signal_received(): data_packet = await queue.get() sync_data_analysis(data_packet) queue.task_done()  My question is if/how it is possible to unit test receive_data and analyse_data. The main logic for reading the data (sync_receive_data_packets) and executing the analysis (sync_data_analysis) are in non-async functions for the sake of simplicity. They are successfully tested. I am not sure how to test the await queue.put and await queue.get() parts especially when they are in endless loops. In fact I am wondering if these functions can be unit tested as they don't return anything. One puts an item into a queue and the other reads it.

Answer: Both functions are not "endless" - the first will end when the generator returned by sync_receive_data_packets is exhausted, and the other whenever termination_signal_received() returns True. Just use mock.patch to point these two functions to callables under the control of your tests. Your tests should further verify if the queue is filled for the first function, and if it is consumed and sync_data_analisys is called with the proper values (it should be patched as well) Otherwise (if they were being repeated with a while True, with no stopping conditions), the way to unit test these would be to write a specialized Queue class that would work as a queue, but that could raise an exception under control of the test code when .get or .put are called.
Question: I am trying to get a long-lived access token from the Instagram graph API through a python script. No matter what I do, I keep receiving the following error: {"error":{"message":"Missing client_id parameter.","type":"OAuthException","code":101,"fbtrace_id":"AogeGpzZGW9AwqCYKHlmKM"}}  My script is the following: import requests refresh_token_url = 'https://graph.facebook.com/v19.0/oauth/access_token' payload = { 'grant_type': 'fb_exchange_token', 'client_id': '1234567890123456' 'client_secret': '1234abcd1234abcd1234abcd', 'fb_exchange_token':'abcd1234' } r = requests.get(refresh_token_url, data=payload)  I am for sure passing the client_id parameter. I verified also by printing r.request.body On the contrary, the request is successful if I send it with the Postman app:  Why do I keep getting that error message? Thank you!

Answer: The Postman called by GET parameters. In modified version, the params argument is used instead of data. This change will append the parameters in payload to the URL as a query string, So your code needs to change it import requests refresh_token_url = 'https://graph.facebook.com/v19.0/oauth/access_token' payload = { 'grant_type': 'fb_exchange_token', 'client_id': '1234567890123456', 'client_secret': '1234abcd1234abcd1234abcd', 'fb_exchange_token': 'abcd1234' } r = requests.get(refresh_token_url, params=payload)
Question: I am wondering if there is a way to type and format strings in Matlab like how you have f-strings in Python: Python input: username = lemslems text = f"Hello {username}!"  Output: Hello lemslems!  I am asking this because I hate writing in the % format and it is just super convenient to write it like how I always do in Python

Answer: Matlab doesn't have a direct equivalent to Python's f-strings, but you can achieve similar functionality using string formatting or concatenation. One way to do this is by using the sprintf function, which formats data into strings: name = 'John'; age = 30; formatted_string = sprintf('My name is %s and I am %d years old.', name, age); disp(formatted_string);  Alternatively, you can use string concatenation: name = 'John'; age = 30; formatted_string = ['My name is ', name, ' and I am ', num2str(age), ' years old.']; disp(formatted_string);  Both approaches will produce the following output: My name is John and I am 30 years old.  These methods allow you to insert variables into strings similar to f-strings in Python.
Question: I am developing a python app and I'm working with pandas dataframe. Unfortunately, I have this warning:  "Setting an item of incompatible dtype is deprecated and will raise an error in a future version of pandas. Value '[149.7 149.7 149.7 149.7]' has dtype incompatible with float64, please explicitly cast to a compatible dtype first. df1.loc[tmp_idx_start:tmp_idx_start+tmp_idx-1, 'AuM'] = (feesPrefinancing.loc[:tmp_idx-1, 'Fees'] + feesMobiDeferred.loc[:tmp_idx-1, 'Fees'] + feesLTQDeferred.loc[:tmp_idx-1, 'Fees']).values "  And here is my code : df1['Date'] = pd.to_datetime(pd.date_range(simulationDate, end=feesLTQDeferred['Valuation date'].iloc[-1], freq='MS', inclusive='both').date) df1['AuM'] = 0. df1.loc[4:7, 'AuM'] = (feesPrefinancing.loc[:3, 'Fees'] + feesMobiDeferred.loc[:3, 'Fees'] + feesLTQDeferred.loc[:3, 'Fees']).values df1.loc[8:, 'AuM'] = (annuities['Invested annuity'].values + feesPrefinancing.loc[4:, 'Fees'].values + feesMobiDeferred.loc[4:, 'Fees'].values + feesLTQDeferred.loc[4:, 'Fees'].values)  As you can guess, I have a dataframe with the first columns being dates and then the second one should contain floats number. That is why I initialized this column with a "0." value (That is also what I read on other post to solve this error). Unfortunately, this warning won't go away even by casting the lines with some "astype" and other things like this. Here is the result obtain which is correct but I have the warning:  Date AuM 0 2023-01-01 0.000000 1 2023-02-01 0.000000 2 2023-03-01 0.000000 3 2023-04-01 0.000000 4 2023-05-01 0.000000 5 2023-06-01 0.000000 6 2023-07-01 0.000000 7 2023-08-01 0.000000 8 2023-09-01 149.700000 9 2023-10-01 149.700000 10 2023-11-01 149.700000 11 2023-12-01 149.700000 12 2024-01-01 2475.800000 13 2024-02-01 2475.800000 14 2024-03-01 2475.800000 15 2024-04-01 2475.800000 16 2024-05-01 2475.800000 17 2024-06-01 2475.800000 18 2024-07-01 2475.800000 19 2024-08-01 2475.800000  Do you have an idea how to solve this ?

Answer: You assigning data types that are not strictly floats to AuM that is set as a float when initialized with 0. Assign the correct data type to AuM: import numpy as np import pandas as pd df1['Date'] = pd.to_datetime(pd.date_range(simulationDate, end=fees:TQDeferred['Valuation date'].iloc[-1], freq='MS', inclusive='both').date) df1['AuM'] = 0.0 df1.loc[4:7, 'AuM'] = (feesPrefinancing.loc[:3, 'Fees'] + feesMoboDeferred.loc[:3, 'Fees'] + feesLTQDeferred.loc[:3, 'Fees']).astype(np.float64).values df1.loc[8:, 'AuM'] = (annuities['Invested annuity'] + feesPrefinancing.loc[4:, 'Fees'] + feesMobiDeferred.loc[4:, 'Fees'] + feesLTQDeferred.loc[4:, 'Fees']).astype(np.float64).values   Chech data types of columns involved in the calculus and make sure they're all numeric. Make sure they are all consistently float64
Question: I want Typer to display my commands in the order I have initialized them and it displays those commands in alphabetic order. I have tried different approaches including this one: https://github.com/tiangolo/typer/issues/246 In this I get AssertionError. Others like subclassing some Typer and click classes does actually nothing. I want the commands to be in the same order as in this working piece of code: import typer import os app = typer.Typer() @app.command() def change_value(file_name, field): print("Here I will change the", file_name, field) @app.command() def close_field(file_name, field): print("I will close field") @app.command() def add_transaction(file_name): print("I will add the transaction") if __name__ == "__main__": app()  Please help :)

Answer: All right I have found solution elsewhere, while trying to post the ticket to Typer (they have a really comprehensive troubleshooting!) https://github.com/tiangolo/typer/issues/428 The app that displays commands in the order I want them displayed looks like this: import typer from click import Context from typer.core import TyperGroup class OrderCommands(TyperGroup): def list_commands(self, ctx: Context): return list(self.commands) app = typer.Typer( cls=OrderCommands, no_args_is_help=True ) @app.command() def change_value(file_name, field): print("Here I will change the", file_name, field) @app.command() def close_field(file_name, field): print("I will close field") @app.command() def add_transaction(file_name): print("I will add the transaction") if __name__ == "__main__": app()
Question: I've been trying to take a DataFrame like df below, and turning some of the columns (say B_m and B_n) into two columns (call them B_m1, B_m2, B_n1 and B_n2), for each pair of values in that column (akin to itertools.combinations(col, r=2)), where they share the same group number. So for B_m, the rows where Group is 0, we should get that B_m1 = [-1, -1, 0] and B_m2 = [0, 1, 1]. df = pd.DataFrame( data=[ [0, 4, 7, -1, 0.9], [0, 4, 7, 0, 0.3], [0, 4, 7, 1, 0.2], [1, 3, 3, 1, 0.5], [1, 3, 3, 0, 0.2], [2, 1, 8, 0, 0.6], ], columns=['Group', 'A_x', 'A_y', 'B_m', 'B_n'], ) print(df)  In the case there is only 1 row with a given group, it should be removed. For 4 or more rows with the same group number, we should similarly find all repetitions without repeats. Shown below is what the expected result should look like. expected = pd.DataFrame( data=[ [0, 4, 7, -1, 0, 0.9, 0.3], [0, 4, 7, -1, 1, 0.9, 0.2], [0, 4, 7, 0, 1, 0.3, 0.2], [1, 3, 3, 1, 0, 0.5, 0.2], ], columns=['Group', 'A_x', 'A_y', 'B_m1', 'B_m2', 'B_n1', 'B_n2'], ) print(expected)  My first attempt not only took ages to run, using some awkard looping and making a new DataFrame (not elegant and don't have it saved), but also didn't work. Since then, I've not been able to come up with an alternative solution. For the record, I did post a similar, but different question about 3 months ago, if that is of any inspiration.

Answer: Here is an approach with groupby.apply and concat: from itertools import combinations cols = ['B_m', 'B_n'] group = list(df.columns.difference(cols, sort=False)) g = df.groupby(group, sort=False) out = pd.concat((g[c].apply(lambda x: pd.DataFrame(list(combinations(x, r=2)))) .rename(columns=lambda x: f'{c}{x+1}') for c in cols), axis=1).reset_index(group)  Output:  Group A_x A_y B_m1 B_m2 B_n1 B_n2 0 1 4 7 -1.0 0.0 0.9 0.3 1 1 4 7 -1.0 1.0 0.9 0.2 2 1 4 7 0.0 1.0 0.3 0.2 0 2 3 3 1.0 0.0 0.5 0.2
Question: I have just installed a package in my virtual environment which I shouldn't have installed. It also installed many dependency packages. Is there a way to roll back and uninstall the package and its dependencies just installed? Something like "uninstall packages installed in the last 1 hour" or a similar functionality.

Answer: Below snippet can help you identify the recently installed pacakage. Once you have the list of packages, you can manually uninstall that pacakge. import pkg_resources import os import time for package in pkg_resources.working_set: print("%s: %s" % (package, time.ctime(os.path.getctime(package.location))))
Question: In Python 3.11 or newer, is there a more convenient type annotation to use than bytes | bytearray for a function argument that means "An ordered collection of bytes"? It seems wasteful to require constructing a bytes from a bytearray (or the other way around) just to satisfy the type-checker. Note that the function does not mutate the argument; it's simply convenient to pass bytes or bytearray instances from different call sites. e.g. def serialize_to_stream(stream: MyStream, data: bytes | bytearray) -> None: for byte in data: stream.accumulate(byte)  (This example is contrived, of course, but the purpose is to show that data is only read, never mutated).

Answer: The typing module used to have a type to represent this: ByteString. However, it was deprecated in 3.9. From the same section:  Prefer collections.abc.Buffer, or a union like bytes | bytearray | memoryview.
Question: I have read similar questions relating to passing JSON text as a command line argument with python, but none of the solutions have worked with my case. I am automating a python script, and the automation runs a powershell script that takes a JSON object generated from a power automate flow. Everything works great until it comes to processing the JSON in my python script. My goal is to convert the JSON to a dictionary so that I can use the key value pairs in my code. My powershell script looks like this:  Python script.py {"Items":[{"Name":"foo","File":"\\\\files\\foo\\foo.csv"},{"Name":"bar","File":"\\\\files\\bar\\bar.csv"},{"Name":"baz","File":"\\\\files\\baz\\baz.csv"}]}  My JSON looks like this: { "Items": [ { "Name": "foo", "File": "\\\\files\\foo\\foo.csv" }, { "Name": "bar", "File": "\\\\files\\bar\\bar.csv" }, { "Name": "baz", "File": "\\\\files\\baz\\baz.csv" } ] }  I tried this solution from SO: if len(sys.argv) > 1: d = json.loads(sys.argv[1]) print(d)  but it returns this error: Unexpected token ':' in expression or statement. + CategoryInfo : ParserError: (:) [], ParentContainsErrorRecordException + FullyQualifiedErrorId : UnexpectedToken  I am unsure how to solve this problem, any suggestions would help!

Answer: Depending on the PowerShell version you're using, you will need to handle it differently, if using pwsh 7.3+ the solution is as simple as wrapping your Json string with single-quotes. Otherwise, if using below that, you would need to escape the double-quotes with \, otherwise those get eaten when passed thru your Python script. Using this can handle it dynamically: $json = @' { "Items": [ { "Name": "foo", "File": "\\\\files\\foo\\foo.csv" }, { "Name": "bar", "File": "\\\\files\\bar\\bar.csv" }, { "Name": "baz", "File": "\\\\files\\baz\\baz.csv" } ] } '@ if ($PSVersionTable.PSVersion -ge '7.3') { Python script.py $json } else { Python script.py $json.Replace('"', '\"') }  Then assuming your Python code would be: import json import sys if len(sys.argv) > 1: d = json.loads(sys.argv[1]) for i in d['Items']: print(i)  Then the output in your PowerShell console would be: {'Name': 'foo', 'File': '\\\\files\\foo\\foo.csv'} {'Name': 'bar', 'File': '\\\\files\\bar\\bar.csv'} {'Name': 'baz', 'File': '\\\\files\\baz\\baz.csv'}   An easier and probably more robust alternative that doesn't require escaping of quotes in any version of PowerShell can be to convert the Json string to Base64 in PowerShell and pass it as argument to your Python script: # Json defined here $json = @' ... ... '@ $b64 = [System.Convert]::ToBase64String([System.Text.Encoding]::UTF8.GetBytes($json)) Python script.py $b64  Then in Python the code works with a minimal change: from base64 import b64decode import json import sys if len(sys.argv) > 1: d = json.loads(b64decode(sys.argv[1])) for i in d['Items']: print(i)