Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- def update_input(self, file, input_type: str, validator) -> bool:
- if (
- instance := self.type_instance.get(input_type)
- ):
- repositories = self.repositories_instance.get(input_type)
- instance = instance.objects.filter(
- scenario=self.scenario,
- initial_file_name=input_type,
- ).first()
- if instance:
- instance.data_file = file
- instance.data_uploaded = True
- instance.save()
- data_frame = pd.read_csv(
- instance.data_file.path,
- )
- data_frame.columns = repositories.get("fields")
- data_frame['id'] = [uuid.uuid4() for _ in range(len(data_frame.index))]
- data_frame["scenario_id"] = data_frame['art_code_nm'].map(lambda _: self.scenario.id)
- # data_frame["stock"] = [int(data_frame['stock'][x]) for x in range(len(data_frame['stock'].index))]
- # print(len(data_frame['stock'].index), len(data_frame.index))
- if not data_frame.empty:
- with get_session() as session:
- data_creator = repositories.get("creator")(
- session=session,
- df=data_frame,
- result_id=instance.id,
- )
- data_creator.execute()
- return True
- return False
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement