eshangj/stackoverflow_q_and_a_sample huggingface dataset:
Question: Are there maintained mypy types published for AWS Lambda functions? I'm talking here about defining a function to handle an HTTP request (from a lambda URL or API Gateway). I'm not talking about other AWS Lambda-related APIs such as are exposed for example by boto3. Here's a example of the kind of function I would like to have types for, taken from AWS's docs: def lambda_handler(event, context): message = 'Hello {} {}!'.format(event['first_name'], event['last_name']) return { 'message' : message }  I'd like to type functions like that one as much as possible - for example, the context parameter (please ignore the fact that this trivial example does not happen to use parameter context). I'm aware that AWS schema registry provides "bindings" for Event Bridge events, but note this question is about invoking lambdas in response to HTTP requests, not Event Bridge events. I'm also aware that for example, if I used FastAPI together with a library like Mangum, I'd get types. I'm restricting this question just to use of the "native" AWS Lambda API (see the example code above). I'm also aware that I could define my own types, but here I'm asking about types maintained by AWS or a third party.

Answer: Check out Powertools for AWS Lambda. It includes typings for LambdaContext as well as dataclasses for event sources such as ALBEvent. from aws_lambda_powertools.utilities.data_classes import ALBEvent from aws_lambda_powertools.utilities.typing import LambdaContext @event_source(data_class=ALBEvent) def lambda_handler(lambda_event: ALBEvent, context: LambdaContext) -> dict[str, Any]: ...
Question: This is part of a larger program I'm working on, but I've got it pinpointed down to the exact problem. When I use Python requests module in some environments it works but not others and it seems to be related to the cipher suite being used by SSL. When I run the following command in Python, it gives an error or success depending on the OS I'm using. here is the command: requests.get("https://api-mte.itespp.org/markets/VirtualService/v2/?WSDL") On Windows 10 and Ubuntu 18.04, it works. But on Windows 11 and Ubuntu 22.04, it does not work and gives the following error: raise SSLError(e, request=request) requests.exceptions.SSLError: HTTPSConnectionPool(host='api-mte.itespp.org', port=443): Max retries exceeded with url: /markets/VirtualService/v2/?WSDL (Caused by SSLError(SSLError(1, '[SSL: DH_KEY_TOO_SMALL] dh key too small (_ssl.c:997)'))) The part [SSL: DH_KEY_TOO_SMALL] dh key too small seems to be the relevant part indicating the requests module thinks the server is using an outdated cipher? When I view the site in Firefox it loads fine. Am I seeing this correctly or am I off base? This seems to be a relevant question (Python - requests.exceptions.SSLError - dh key too small) but I could not get it to work when I changed  requests.packages.urllib3.util.ssl_.DEFAULT_CIPHERS. I know Ubuntu 22 uses OpenSSL v3 while Ubuntu 18 uses OpenSSL v1.1.1 and that seems relevant. Thanks! EDIT: I believe specifically I'm trying to use the cipher ECDHE-RSA-AES128-GCM-SHA256 or ECDHE-RSA-AES256-GCM-SHA384

Answer: So indeed the problem is the server is using an outdated cipher. And the issue is that Windows 11 and Ubuntu 22 will not use that outdated cipher by default. The issue also is that in the version of requests that I was using (2.31.0), you can no longer change the cipher suite with requests.packages.urllib3.util.ssl_.DEFAULT_CIPHERS as many answers suggest. The answer was found here, a blog post written by I believe one of the authors of the module. It involves making a new adapter from the requests module so the solution isn't straightforward. This was the code I used to make it work. Obviously you can change custom_ciphers to whatever is needed. import requests from requests.adapters import HTTPAdapter from requests.packages.urllib3.util.ssl_ import create_urllib3_context import urllib3 custom_ciphers = [ "ECDHE-RSA-AES256-GCM-SHA384", # "DHE-RSA-AES256-GCM-SHA384", # "ECDHE-RSA-AES128-GCM-SHA256", # "TLS_AES_256_GCM_SHA384", ] class CustomCipherAdapter(HTTPAdapter): def init_poolmanager(self, *args, **kwargs): context = create_urllib3_context(ciphers=":".join(custom_ciphers)) kwargs['ssl_context'] = context return super(CustomCipherAdapter, self).init_poolmanager(*args, **kwargs) # Create a session and mount the adapter session = requests.Session() session.mount("https://", CustomCipherAdapter()) # Now you can use the session to make requests with the custom cipher suites response = session.get("https://api-mte.itespp.org/markets/VirtualService/v2/?WSDL") print(response)  Info about HTTPTransport can be found here.
Question: I want to create a turtle game. When I click the turtle shape in the game, it should give score +1 point so that I can count the points. But when I execute the command and click on the turtle shape it is giving me a name error which is NameError: name 'score_turtle' is not defined even though it is defined in the upper command. Please help me to fix it. import turtle import random screen = turtle.Screen() screen.bgcolor("light blue") FONT = ('Arial', 25, 'normal') skor = 0 turtle_list = [] #score turtle def score_turtle_setup(): score_turtle = turtle.Turtle() score_turtle.hideturtle() score_turtle.penup() score_turtle.color("dark blue") top_screen = screen.window_height() / 2 y = top_screen * 0.85 score_turtle.setpos(-200,y) score_turtle.write(arg=f"SCORE: 0", move=False, align="center", font=FONT) def make_turtle(x, y): point = turtle.Turtle() def handle_click(x, y): global skor skor += 1 score_turtle.write(arg=f"SCORE: {skor}", move=False, align="center", font=FONT) point.onclick(handle_click) point.shape("turtle") point.penup() point.color("green") point.shapesize(1.5) point.goto(x * 12, y*12) turtle_list.append(point) x_coordinats = [-20,-10,0,10,20] y_coordinats = [20,10,0,-10] def setup_turtles(): for x in x_coordinats: for y in y_coordinats: make_turtle(x, y) def hide_turtles(): for point in turtle_list: point.hideturtle() def show_turtles(): random.choice(turtle_list).showturtle() turtle.tracer(0) score_turtle_setup() setup_turtles() hide_turtles() show_turtles() turtle.tracer(1) turtle.mainloop()  I tried to call the function score_turtle_setup from the handle_click function but that did not work.

Answer: Variables are scoped to the function by default. That means that when the function returns, all variables in the function scope are destroyed. Luckily, functions let you return values, so you can do something like: def score_turtle_setup(): score_turtle = turtle.Turtle() # ... return score_turtle # ... score_turtle = score_turtle_setup()  Now the score_turtle variable is assigned in the global scope, where it can be accessed from any of your other functions. This may not be great design, but for a small program, it's fine. Down the line, if the program grows, consider using a class and/or module for encapsulating logical entities like this. In general, the situation isn't specific to turtle, so How do I get a result (output) from a function? How can I use the result later? is a good canonical resource.
Question: Say I want to have a function exec_myfunc that can execute any user-defined function with an input value of 10. The user is supposed to define the function by a string as shown below: func1_str = """ def myfunc1(x): return x """ func2_str = """ def myfunc2(x): return x**2 """  Right now I am using a very hacky way, by extracting the function name between def  and ( using regular expression, as shown below: def exec_myfunc(func_str: str): import re exec(func_str) myfunc_str = re.search(r'def(.*)\(', func_str).group(1).strip() return eval(myfunc_str)(10) print(exec_myfunc(func1_str)) # 10 print(exec_myfunc(func2_str)) # 100  I would like to know what is the general and correct way of doing this?

Answer: You could directly use exec: def exec_myfunc(func_str: str): fun = {} # to hold your function exec(func_str,fun) fun_name = list(fun.keys())[1] fn = fun[fun_name] print(f"The function {fun_name} evaluated at 10 = {fn(10)}") exec_myfunc(func1_str) The function myfunc1 evaluated at 10 = 10
Question: I need like a magnet effect to form a cluster within a list of unique items. My input is this : lst1 = ['ah', 'ab', 'c', None, 'xo', 'i', 'b', 'lji', 'z', 'bel', 'oyb']  So based on a given item (a string that could be of any length), we need to move from left and right (towards this item) every string that contains it. For simplicity, lets consider the chosen item as a single letter : letter = 'b'. Now, we need to look for any string that contain b and bring it closer to it (from both sides) and preserve its suborder, like magnet. The other items will also be shifted (-+). In this case, the output should be : lst2 = ['ah', 'c', None, 'xo', 'i', 'ab', 'b', 'bel', 'oyb', 'lji', 'z']  The cluster is represented here by the slice : ['ab', 'b', 'bel', 'oyb']. I tried the code below but I got a wrong list even though I was able to make the cluster : letter = 'b' data = {} for i, e in enumerate(lst1): if e is not None and letter in e and e!=letter: data[e] = i left = lst1.index(letter)-1 right = lst1.index(letter)+1 for key,value in data.items(): if value<lst1.index(letter): lst1[left] = key left -= 1 elif value>lst1.index(letter): lst1[right] = key right += 1  Here is what I got : print(lst1) ['ah', 'ab', 'c', None, 'xo', 'ab', 'b', 'bel', 'oyb', 'bel', 'oyb']  Do you have any suggestions ?

Answer: One possible solution is to use list slicing + sorting: letter = "b" lst1 = ["ah", "ab", "c", None, "xo", "i", "b", "lji", "z", "bel", "oyb"] index = lst1.index(letter) out = ( sorted(lst1[:index], key=lambda v: letter in v if isinstance(v, str) else False) + [letter] + sorted(lst1[index + 1 :], key=lambda v: letter not in v if isinstance(v, str) else True) ) print(out)  Prints: ['ah', 'c', None, 'xo', 'i', 'ab', 'b', 'bel', 'oyb', 'lji', 'z']
Question: Given the following table, I'd like to remove the duplicates based on the column subset col1,col2. I'd like to keep the first row of the duplicates though: data = { 'col1': [1, 2, 3, 1, 1], 'col2': [7, 8, 9, 7, 7], 'col3': [3, 4, 5, 6, 8] } tmp = pl.DataFrame(data)  ┌──────┬──────┬──────┐ │ col1 ┆ col2 ┆ col3 │ │ --- ┆ --- ┆ --- │ │ i64 ┆ i64 ┆ i64 │ ╞══════╪══════╪══════╡ │ 1 ┆ 7 ┆ 3 │ │ 2 ┆ 8 ┆ 4 │ │ 3 ┆ 9 ┆ 5 │ │ 1 ┆ 7 ┆ 6 │ │ 1 ┆ 7 ┆ 9 │ └──────┴──────┴──────┘  The result should be ┌──────┬──────┬──────┐ │ col1 ┆ col2 ┆ col3 │ │ --- ┆ --- ┆ --- │ │ i64 ┆ i64 ┆ i64 │ ╞══════╪══════╪══════╡ │ 1 ┆ 7 ┆ 3 │ │ 2 ┆ 8 ┆ 4 │ │ 3 ┆ 9 ┆ 5 │ └──────┴──────┴──────┘  Usually I'd do this with pandas df["col1","col2"].is_duplicated(keep='first'), but polars dl.is_duplicated() marks all rows as duplicates including the first occurence.

Answer: You can use DataFrame.unique; the flexible keep keyword argument is here in Polars. tmp.unique(('col1', 'col2'), keep='first', maintain_order=True)
Question: I'm trying to build my code using CloudBuild, but it's failing with the following message: Error: Error while updating cloudfunction configuration: Error waiting for Updating CloudFunctions Function: Error code 3, message: Build failed: found incompatible dependencies: "functions-framework 3.0.0 has requirement flask<3.0,>=1.0, but you have flask 3.0.0."; Error ID: 5503c41a │  I've checked my dependencies and subdependencies, and I've tried to release the same code that was releasable 2 weeks ago, and I'm getting this error everywhere. I'm not using Flask 3.0.0 anywhere, my version is 2.3.3. My guess is that this is some temporary problem, since I haven't found anything online, and google has already had problems with functions-framework before, but I'm interested if someone has found a workaround?

Answer: Solved by setting fixed (latest) version of functions-framework, which is 3.4.0 at the time of writing this comment in my requirements.txt. Why GCP doesn't use latest version of their own libary is beyond me.
Question: I am trying to print the datastream by applying the tumbling process window for every 5 seconds. Since I couldn't implement the custom deserializer for now, I created the process function which returns the result as tuple, and as per this documentation link I could link the process function with the windowing operation so I tried this out: def get_data(self): source = self.__get_kafka_source() ds = self.env.from_source(source, WatermarkStrategy.no_watermarks(), "Kafka Source").window_all( TumblingProcessingTimeWindows.of(Time.seconds(5))) ds.process(ExtractingRecordAttributes(), output_type=Types.TUPLE( [Types.STRING(), Types.STRING(), Types.STRING()])).print() self.env.execute("source") def __get_kafka_source(self): source = KafkaSource.builder() \ .set_bootstrap_servers("localhost:9092") \ .set_topics("test-topic1") \ .set_group_id("my-group") \ .set_starting_offsets(KafkaOffsetsInitializer.latest()) \ .set_value_only_deserializer(SimpleStringSchema()) \ .build() return source class ExtractingRecordAttributes(KeyedProcessFunction): def __init__(self): pass def process_element(self, value: str, ctx: 'KeyedProcessFunction.Context'): parts = UserData(*ast.literal_eval(value)) result = (parts.user, parts.rank, str(ctx.timestamp())) yield result def on_timer(self, timestamp, ctx: 'KeyedProcessFunction.OnTimerContext'): yield "On timer timestamp: " + str(timestamp)  When I trigger the get_data method, it gives me the below error:  return self._wrapped_function.process(self._internal_context, input_data) AttributeError: 'ExtractingRecordAttributes' object has no attribute 'process' at java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:395) at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1999) at org.apache.beam.sdk.util.MoreFutures.get(MoreFutures.java:61) at org.apache.beam.runners.fnexecution.control.SdkHarnessClient$BundleProcessor$ActiveBundle.close(SdkHarnessClient.java:504) at org.apache.beam.runners.fnexecution.control.DefaultJobBundleFactory$SimpleStageBundleFactory$1.close(DefaultJobBundleFactory.java:555) at org.apache.flink.streaming.api.runners.python.beam.BeamPythonFunctionRunner.finishBundle(BeamPythonFunctionRunner.java:421) ... 7 more  If I don't use window_all, everything works fine. But the moment I introduce it, it fails. What am I doing wrong here? Any hints would be helpful. I am using Pyflink 1.17.1 TIA.

Answer: The problem is for windowing you need to use ProcessWindowFunction and not ProcessFunction. Ideally, your code should look like this: class ExtractingRecordAttributes(ProcessWindowFunction): def __init__(self): pass def process(self, key: str, context: 'ProcessWindowFunction.Context', elements: Iterable[Tuple[str, str]]) -> Iterable[Tuple[str, str, str]]: result = "" for element in elements: parts = UserData(*ast.literal_eval(str(element))) result = (parts.user, parts.rank, str(context.current_processing_time())) yield result def clear(self, context: 'ProcessWindowFunction.Context'): pass  I haven't tested this, but I think it should work. Here is the reference from the documentation.
Question: I want to get multiple small slices out of a large time-series dataset (~ 25GB, ~800M rows). At the moment, this looks something like this: from polars import pl sample = pl.scan_csv(FILENAME, new_columns=["time", "force"]).slice(660_000_000, 3000).collect()  This code takes about 0-5 minutes, depending on the position of the slice I want to get. If I want 5 slices, this takes maybe 15 minutes to run everything. However, since polars is reading the whole csv anyhow, I was wondering if there is a way to get all my slices I want in one go, so polars only has to read the csv once. Chaining multiple slices (obviously) doesn't work, maybe there is some other way?

Answer: You should be able to run them all in parallel with .collect_all() lf = pl.scan_csv(FILENAME, new_columns=["time", "force"]) samples = [ lf.slice(660_000_000, 3000), lf.slice(890_000_000, 5000), ... ] samples = pl.collect_all(samples)
Question: My input is this dataframe : df = pd.DataFrame({'class': ['class_a', 'class_a', 'class_a', 'class_a', 'class_b', 'class_b', 'class_c', 'class_c', 'class_d'], 'id': ['id1', 'id2', 'id3', '', '', 'id4', 'id5', '', 'id6'], 'name': ['abc1', '', '', 'abc2', 'abc3', '', '', 'abc4', 'abc5']}) print(df) class id name 0 class_a id1 abc1 1 class_a id2 2 class_a id3 3 class_a abc2 4 class_b abc3 5 class_b id4 6 class_c id5 7 class_c abc4 8 class_d id6 abc5  And I need to have one single row per pair (class/id). We need to move up the values in id and name as long as it's possible.  class id name 0 class_a id1 abc1 1 class_a id2 abc2 2 class_a id3 3 class_b id4 abc3 4 class_c id5 abc4 5 class_d id6 abc5  I tried with the code below but I get a messed up result :  df.replace('',np.nan,inplace=True) df.groupby(['class', 'id'], as_index=False, sort=False).first() class id name 0 class_a id1 abc1 1 class_a id2 None 2 class_a id3 None 3 claas_b id4 None 4 class_c id5 None 5 class_d id6 abc5  Can you provide an explanation please ?

Answer: Use DataFrame.melt with sorting by key parameter for correct ordering, then reshape by DataFrame.pivot with GroupBy.cumcount and last remove empty rows if exist at least one value: out = (df.melt('class').sort_values('value', key=lambda x: x.eq('')) .assign(g = lambda x: x.groupby(['class','variable']).cumcount()) .pivot(index=['class','g'], columns='variable', values='value') .droplevel(1) .reset_index() .rename_axis(None, axis=1)) cols = out.columns.difference(['class']) out = out[out[cols].ne('').any(axis=1) | ~df['class'].duplicated(keep=False)] print (out) class id name 0 class_a id1 abc1 1 class_a id2 abc2 2 class_a id3 4 class_b id4 abc3 6 class_c id5 abc4 8 class_d id6 abc5 9 class_e  Input DataFrame: df = pd.DataFrame({'class': ['class_a', 'class_a', 'class_a', 'class_a', 'class_b', 'class_b', 'class_c', 'class_c', 'class_d', 'class_e'], 'id': ['id1', 'id2', 'id3', '', '', 'id4', 'id5', '', 'id6',''], 'name': ['abc1', '', '', 'abc2', 'abc3', '', '', 'abc4', 'abc5','']}) print(df) class id name 0 class_a id1 abc1 1 class_a id2 2 class_a id3 3 class_a abc2 4 class_b abc3 5 class_b id4 6 class_c id5 7 class_c abc4 8 class_d id6 abc5 9 class_e
Question: I am working with a MultiIndex DataFrame and would like to melt out one of the column levels, I have found a way to do this but it involves a two step process (detailed below). I am wondering if there is a way in which I can use df.melt() to achieve the intended result in one step Here is a simplified example of the df I am working with import pandas as pd import datetime import numpy as np cols = pd.MultiIndex.from_tuples( [ ('A', 'temp_avg'), ('A', 'temp_predicted'), ('B', 'temp_avg'), ('B', 'temp_predicted'), ] ) df = pd.DataFrame( np.random.randn(4,4), columns=cols, index=pd.date_range( datetime.datetime(2013,1,1), datetime.datetime(2013,1,4), ) ) df.index.names=['date']  My current two-step solution is as follows: df_melt = df.reset_index().melt( id_vars=[('date', "")], var_name=['location', 'name'], ).rename({('date', ''): 'date'}, axis=1) df_desired = df_melt.pivot_table( index=['date', 'location'], columns='name', values='value', )  I wish to arrive at df_desired by a one-step use of df.melt(**kwargs) or another function

Answer: You just need to stack: df_desired = df.stack(0)  With name: df_desired = df.rename_axis(['location', None], axis=1).stack('location')  Output:  temp_avg temp_predicted date location 2013-01-01 A 0.018696 -1.135884 B 0.064724 0.790992 2013-01-02 A -1.572779 -0.365371 B -0.572017 0.742684 2013-01-03 A -3.018399 1.081398 B 1.223285 -1.810627 2013-01-04 A -0.889924 -0.652284 B -0.251828 0.098451
Question: I've looked at some examples to mock environment variables. Here In my case, I have to mock a variable in my config.py file that is setup as follows: class ServiceConfig(BaseSettings): ... API_AUDIENCE: str = os.environ.get("API_AUDIENCE") ... ...  I have that called here: class Config(BaseSettings): auth: ClassVar[ServiceConfig] = ServiceConfig()  I've attempted to patch that value as follows: @mock.patch.dict(os.environ, {"API_AUDIENCE": "https://mock.com"}) class Test(TestCase):  Another way I've tried is: @patch('config.Config.API_AUDIENCE', "https://mock.com") def ....:  The setup is incorrect both ways, getting this error: E pydantic_core._pydantic_core.ValidationError: 1 validation error for ServiceConfig E API_AUDIENCE  What is the correct way to setup mocking an environment variable?

Answer: To update the environment, and restore it after the test: import os from unittest import mock @pytest.fixture() def setenvvar(monkeypatch): with mock.patch.dict(os.environ, clear=True): envvars = { "API_AUDIENCE": "https://mock.com", } for k, v in envvars.items(): monkeypatch.setenv(k, v) yield # This is the magical bit which restore the environment after  This is just a fixture, you can use it (or not) in any test you want, at the scope you want, no extra dependencies.
Question: I'm trying to use embedded python in my C++ dll library. The library is built and compiled in MSYS2 using GCC compiler, CMake and Ninja. Python 3.10 is also installed on MSYS2 using pacman. Windows 10 env contains C:\msys64\mingw64\bin in Path (python is also located there). Python doesn't installed on Windows, only on MSYS2. This is what CMakeLists.txt contains: cmake_minimum_required(VERSION 3.26) project(python_test) set(CMAKE_CXX_STANDARD 17) find_package(Python REQUIRED Development) add_executable(python_test main.cpp) target_link_libraries(python_test PRIVATE Python::Python)  Simple test code: int main() { Py_Initialize(); PyRun_SimpleString("from time import time,ctime\n" "import numpy as np\n" "print('Today is', ctime(time()))\n"); if (Py_FinalizeEx() < 0) exit(120); return 0; }  When I run this code I got this error: Could not find platform independent libraries <prefix> Could not find platform dependent libraries <exec_prefix> Consider setting $PYTHONHOME to <prefix>[:<exec_prefix>] Python path configuration: PYTHONHOME = (not set) PYTHONPATH = (not set) program name = 'python3' isolated = 0 environment = 1 user site = 1 import site = 1 sys._base_executable = 'C:\\Users\\someUsername\\CLionProjects\\python_test\\cmake-build-debug\\python_test.exe' sys.base_prefix = 'D:\\a\\msys64\\mingw64' sys.base_exec_prefix = 'D:\\a\\msys64\\mingw64' sys.platlibdir = 'lib' sys.executable = 'C:\\Users\\someUsername\\CLionProjects\\python_test\\cmake-build-debug\\python_test.exe' sys.prefix = 'D:\\a\\msys64\\mingw64' sys.exec_prefix = 'D:\\a\\msys64\\mingw64' sys.path = [ 'D:\\a\\msys64\\mingw64\\lib\\python310.zip', 'D:\\a\\msys64\\mingw64\\lib\\python3.10', 'D:\\a\\msys64\\mingw64\\lib\\lib-dynload', '', ] Fatal Python error: init_fs_encoding: failed to get the Python codec of the filesystem encoding Python runtime state: core initialized ModuleNotFoundError: No module named 'encodings' Current thread 0x00004564 (most recent call first): <no Python frame>  But python command work both in MSYS2/MinGW console and Windows cmd. When I specify PYTHONHOME to C:\msys64\mingw64\bin in Windows 10 environment, I have got the same error again in Clion, and now in Windows command line too. How to resolve this problem?

Answer: I set PYTHONHOME=C:\msys64\mingw64\bin and more important PYTHONPATH=C:\msys64\mingw64\lib\python3.10;C:\msys64\mingw64\lib\python3.10\site-packages;C:\msys64\mingw64\lib\lib-dynload and it solved my problem