1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

fromdjango.db importmodels

# Create your models here.

# 上传文件表

classAttachment(models.Model):

    id=models.AutoField(primary_key=True, verbose_name='自增id')

    name =models.CharField(max_length=200, verbose_name='附件名称')

    file_path =models.CharField(max_length=200, verbose_name='附件相对路径')

    create_time =models.DateTimeField(verbose_name='上传时间')

    classMeta:

    db_table ='tb_attachment'

    verbose_name ='附件表'

    verbose_name_plural =verbose_name

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

#!/usr/bin/env python

# -*- coding:utf-8 -*-

__author__ ='授客'

fromdjango.contrib importadmin

fromdjango.urls importpath

fromdjango.conf.urls importinclude

urlpatterns =[

path('admin/', admin.site.urls),

path('', include('mywebsite.urls')) #添加API路由配置（这里根据项目实际情况配置）

]

1

2

MEDIA_URL ='/media/'

MEDIA_ROOT =os.path.join(BASE_DIR, 'media').replace('\\', '/')

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

#!/usr/bin/env python

# -*- coding:utf-8 -*-

__author__ ='授客'

fromrest_framework.views importAPIView

fromrest_framework.response importResponse

fromrest_framework importstatus

from.models importAttachment

fromdjango.http importFileResponse

fromdjango.utils importtimezone

fromdjango.conf importsettings

importos

importuuid

importlogging

logger =logging.getLogger('mylogger')

# 批量创建目录

defmkdirs_in_batch(path):

    try:

        path =os.path.normpath(path)  # 去掉路径最右侧的 \\ 、/

        path =path.replace('\\', '/') # 将所有的\\转为/，避免出现转义字符串

        head, tail =os.path.split(path)

        ifnotos.path.isdir(path) andos.path.isfile(path):  # 如果path指向的是文件，则分解文件所在目录

            head, tail =os.path.split(head)

        iftail =='': # head为根目录，形如 / 、D:

            returnTrue

        new_dir_path =''  # 存放反转后的目录路径

        root =''  # 存放根目录

        whiletail:

            new_dir_path =new_dir_path +tail +'/'

            head, tail =os.path.split(head)

            root =head

        else:

            new_dir_path =root +new_dir_path

            # 批量创建目录

            new_dir_path =os.path.normpath(new_dir_path)

            head, tail =os.path.split(new_dir_path)

            temp =''

            whiletail:

                temp =temp +'/'+tail

                dir_path =root +temp

                ifnotos.path.isdir(dir_path):

                    os.mkdir(dir_path)

                head, tail =os.path.split(head)

        returnTrue

    exceptException as e:

        logger.error('批量创建目录出错：%s'%e)

        returnFalse

classAttachmentAPIView(APIView):

# 上传附件

defpost(self, request, format=None):

    result ={}

    try:

        files =request.FILES

        file=files.get('file')

        ifnotfile:

            result['msg'] ='上传失败，未获取到文件'

            result['success'] =False

            returnResponse(result, status.HTTP_400_BAD_REQUEST)

        # data = request.POST #获取前端发送的，file之外的其它参数

        # extra = data.get('extra')

        file_name =file.name

        attachment_name =file_name

        creater =request.user.username

        create_time =timezone.now()

        time_str =create_time.strftime('%Y%m%d')

        name, suffix =os.path.splitext(file_name)

        file_name =str(uuid.uuid1()).replace('-', '') +time_str +suffix

        file_relative_path ='/myapp/attachments/'+time_str

        file_absolute_path =settings.MEDIA_ROOT +file_relative_path

        ifnotos.path.exists(file_absolute_path):# 路径不存在

            ifnotutils.mkdirs_in_batch(file_absolute_path):

                result['msg'] ='批量创建路径(%s)对应的目录失败'%file_absolute_path

                result['success'] =False

                returnResponse(result, status.HTTP_500_INTERNAL_SERVER_ERROR)

        file_relative_path +='/'+file_name

        data['file_path'] =file_relative_path

        file_absolute_path =file_absolute_path +'/'+file_name

        file_handler =open(file_absolute_path, 'wb')    # 打开特定的文件进行二进制的写操作

        try:

            forchunk infile.chunks():      # 分块写入文件

                file_handler.write(chunk)

        finally:

            file_handler.close()

        # 记录到数据库

        try:

            obj =Attachment(file_path=file_path, name=attachment_name, create_time=create_time, creater=creater)

            obj.save()

        exceptException as e:

            result['msg'] ='上传失败：%s'%e

            result['success'] =False

            returnResponse(result, status.HTTP_400_BAD_REQUEST)

        result['msg'] ='上传成功'

        result['success'] =True

        result['data'] =result_data

        returnResponse(result, status.HTTP_200_OK)

    exceptException as e:

        result['msg'] ='%s'%e

        result['success'] =False

        returnResponse(result, status.HTTP_500_INTERNAL_SERVER_ERROR)

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

# 删除附件

defdelete(self, request, format=None):

    result ={}

    try:

        data =request.data

        attachment_id =data.get('attachment_id')

        obj =Attachment.objects.filter(id=attachment_id).first()

        ifobj:

            file_absoulte_path =settings.MEDIA_ROOT +'/'+obj.file_path

            ifos.path.exists(file_absoulte_path) andos.path.isfile(file_absoulte_path):

                os.remove(file_absoulte_path)

                obj.delete()

        result['msg'] ='删除成功'

        result['success'] =True

        returnResponse(result, status.HTTP_200_OK)

    exceptException as e:

        result['msg'] ='%s'%e

        result['success'] =False

        returnResponse(result, status.HTTP_500_INTERNAL_SERVER_ERROR)

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

# 下载附件

defget(self, request, format=None):

    result ={}

    try:

        data =request.GET

        attachment_id =data.get('attachmentId')

        obj =Attachment.objects.filter(id=attachment_id).first()

        ifobj:

            file_absoulte_path =settings.MEDIA_ROOT+obj.file_path

            ifos.path.exists(file_absoulte_path) andos.path.isfile(file_absoulte_path):

                file=open(file_absoulte_path, 'rb')

                file_response =FileResponse(file)

                file_response['Content-Type']='application/octet-stream'

                file_response["Access-Control-Expose-Headers"] ='Content-Disposition'# 设置可以作为响应的一部分暴露给外部的请求头，如果缺少这行代码，会导致前端请求响应中看不到该请求头

                file_response['Content-Disposition']='attachment;filename={}'.format(urlquote(obj.name)) # 这里使用urlquote函数主要为针对文件名为中文时，对文件名进行编码，编码后，前端获取的文件名称形如“%E5%AF%BC%E5%87%BA%E6%B5%8B%E8%AF%95%E7%94%A8%E4%BE%8B”

                returnfile_response

            else:

                result['msg'] ='请求失败,资源不存在'

                result['success'] =False

        else:

            result['msg'] ='请求失败,资源不存在'

            result['success'] =False

            returnResponse(result, status.HTTP_200_OK)

    exceptException as e:

        result['msg'] ='%s'%e

        result['success'] =False

        returnResponse(result, status.HTTP_500_INTERNAL_SERVER_ERROR)

1

2

3

4

5

6

7

8

9

10

11

12

13

14

#!/usr/bin/env python

# -*- coding:utf-8 -*-

__author__ ='授客'

fromdjango.urls importre_path

from.views importAttachmentAPIView

urlpatterns =[

#...略

re_path('^api/v1/testcase/\d+/attachment$', testcase_attachment_views.TestcaseAttachmentAPIView.as_view()), # 给测试用例添加附件

re_path('^api/v1/testcase/\d+/attachment/\d+$', testcase_attachment_views.TestcaseAttachmentAPIView.as_view()), # 删除、下载测试用例关联的附件