Skip to content

Commit

Permalink
fix:add typing and reorder functions
Browse files Browse the repository at this point in the history
  • Loading branch information
Abdellahitech committed Jan 9, 2024
1 parent ba68f4a commit 9e851cf
Showing 1 changed file with 18 additions and 16 deletions.
34 changes: 18 additions & 16 deletions src/hrflow_connectors/connectors/talentsoft/warehouse.py
Original file line number Diff line number Diff line change
Expand Up @@ -170,14 +170,16 @@ class WriteProfileParameters(ParametersModel):
)


def decode_unicode(input_str):
def decode_unicode(input_str: str) -> str:
try:
return bytes(input_str, "utf-8").decode("unicode_escape")
except UnicodeDecodeError:
return input_str


def decode_json(obj):
def decode_json(
obj: t.Union[str, list, dict, t.Any]
) -> t.Union[str, list, dict, t.Any]:
if isinstance(obj, str):
return decode_unicode(obj)
elif isinstance(obj, list):
Expand All @@ -188,6 +190,20 @@ def decode_json(obj):
return obj


def get_mime_type_with_mimetypes(filename: t.Optional[str]) -> str:
if filename is None:
return "application/octet-stream"
mime_type, encoding = mimetypes.guess_type(filename)
return mime_type or "application/octet-stream"


def get_cv_content(attachment: dict) -> t.Optional[bytes]:
response = requests.get(attachment["public_url"])
if response.status_code == 200:
return response.content
raise Exception(response.text)


def get_talentsoft_auth_token(
client_url: str,
client_id: str,
Expand Down Expand Up @@ -228,13 +244,6 @@ def get_talentsoft_auth_token(
)


def get_mime_type_with_mimetypes(filename):
if filename is None:
return "application/octet-stream"
mime_type, encoding = mimetypes.guess_type(filename)
return mime_type or "application/octet-stream"


def post_applicant_front(client_url, token, applicant, files, job_reference=None):
if job_reference:
headers = {
Expand All @@ -258,13 +267,6 @@ def post_applicant_front(client_url, token, applicant, files, job_reference=None
raise Exception(response.text)


def get_cv_content(attachment):
response = requests.get(attachment["public_url"])
if response.status_code == 200:
return response.content
raise Exception(response.text)


def read_jobs(
adapter: LoggerAdapter,
parameters: ReadProfilesParameters,
Expand Down

0 comments on commit 9e851cf

Please sign in to comment.