FilesExpand file tree

master

/

auth.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

123 lines (104 loc) · 3.88 KB

master

/

auth.py

Copy path

Top

File metadata and controls

• Code
• Blame

123 lines (104 loc) · 3.88 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

from pydantic import BaseModel

from typing import Union

from fastapi import Depends, HTTPException, status

from fastapi.security import OAuth2PasswordBearer

import psycopg2

import config as c

import db_func

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="sign_in")

users = {

"pupochek": {

"username": "pupochek",

"token": "pupochek",

"position": "student"

},

"secretary2": {

"username": "secretary2",

"token": "secretary2",

"position": "secretary"

}, 'study_department2': {

'username': 'study_department2',

'token': 'study_department2',

'position': 'study_department'},

'trainer2': {

'username': 'trainer2',

'token': 'trainer2',

'position': 'trainer'}

}

class User(BaseModel):

username: str

token: str

position: str

async def get_current_user(token: str = Depends(oauth2_scheme)):

for login, body in users.items():

print(login, body)

if body["token"] == token:

return login

raise HTTPException(

status_code=status.HTTP_401_UNAUTHORIZED,

detail="Invalid authentication credentials",

headers={"WWW-Authenticate": "Bearer"},

)

async def get_current_user_position(token: str = Depends(oauth2_scheme)):

print(users)

return users[token]["position"]

async def get_current_active_user(current_user: User = Depends(get_current_user)):

return current_user

def create_user(position, first_name, surname, middle_name, login,

password, id_student,

institute,

direction, speciality, direction_military, birth_date):

if position == 'student' and db_func.user_exist(login) is False:

db_func.create_student(id_student, first_name, surname, middle_name,

institute,

direction, speciality, direction_military, login,

password, birth_date)

return "student created"

elif position in ['study_department', 'trainer', 'secretary'] and db_func.user_exist(login) is False:

db_func.create_employee(first_name, surname, middle_name, position, login,

password)

return position + " user created"

elif db_func.user_exist(login):

return "User exist"

else:

return "There is no such position: " + position

def token_user_db(login, password):

conn = psycopg2.connect(dbname=c.db, user=c.user,

password=c.password, host=c.host)

cursor = conn.cursor()

cursor.execute(

'SELECT login_candidate FROM candidate WHERE login_candidate = %s AND password_candidate = crypt(%s,password_candidate);',

[login, password])

login_candidate = cursor.fetchone()

cursor.execute(

'SELECT login_employee FROM employee WHERE login_employee = %s AND password_employee = crypt(%s, password_employee);',

[login, password])

login_employee = cursor.fetchone()

print(login_candidate)

print(login_employee)

if login_candidate is not None:

users[login] = {"username": login,

"token": login,

"position": "student"}

print(users)

cursor.close()

conn.close()

return login_candidate

elif login_employee is not None:

print('employee')

cursor.execute(

'SELECT name_position FROM classifier_employee WHERE id_position = (SELECT id_position FROM employee WHERE login_employee = %s);',

[login])

position = cursor.fetchone()

print(position)

users[login] = {"username": login,

"token": login,

"position": position[0]}

print(users)

cursor.close()

conn.close()

return login_employee

else:

cursor.close()

conn.close()

return 'No such user'