Refactoring. Using black code style

Author: gil9red

socket__tcp__examples/common.py

@@ -1,15 +1,14 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 
-__author__ = 'ipetrash'
+__author__ = "ipetrash"
 
 
 # SOURCE: https://stackoverflow.com/a/17668009/5909792
 
 
 import struct
 import zlib
-from typing import Optional
 
 
 def crc32_from_bytes(data: bytes) -> int:
@@ -19,12 +18,12 @@ def crc32_from_bytes(data: bytes) -> int:
 def send_msg__with_crc32(sock, msg):
     # Prefix each message with a 12-byte (network byte order) length: 8-byte data length and 4-byte crc32 data
     crc32 = crc32_from_bytes(msg)
-    msg = struct.pack('>QI', len(msg), crc32) + msg
+    msg = struct.pack(">QI", len(msg), crc32) + msg
 
     sock.sendall(msg)
 
 
-def recv_msg__with_crc32(sock) -> Optional[bytes]:
+def recv_msg__with_crc32(sock) -> bytes | None:
     # 12-byte
     payload_size = struct.calcsize(">QI")
 
@@ -33,26 +32,30 @@ def recv_msg__with_crc32(sock) -> Optional[bytes]:
     if not raw_msg_len:
         return None
 
-    msg_len, crc32 = struct.unpack('>QI', raw_msg_len)
+    msg_len, crc32 = struct.unpack(">QI", raw_msg_len)
 
     # Read the message data
     msg = recv_all(sock, msg_len)
 
     # Check message
     msg_crc32 = crc32_from_bytes(msg)
     if msg_crc32 != crc32:
-        raise Exception('Incorrect message: invalid crc32. Receiving crc32: {}, current: {}'.format(crc32, msg_crc32))
+        raise Exception(
+            "Incorrect message: invalid crc32. Receiving crc32: {}, current: {}".format(
+                crc32, msg_crc32
+            )
+        )
 
     return msg
 
 
 def send_msg(sock, msg):
     # Prefix each message with a 8-byte length (network byte order)
-    msg = struct.pack('>Q', len(msg)) + msg
+    msg = struct.pack(">Q", len(msg)) + msg
     sock.sendall(msg)
 
 
-def recv_msg(sock) -> Optional[bytes]:
+def recv_msg(sock) -> bytes | None:
     # 8-byte
     payload_size = struct.calcsize(">Q")
 
@@ -61,13 +64,13 @@ def recv_msg(sock) -> Optional[bytes]:
     if not raw_msg_len:
         return None
 
-    msg_len = struct.unpack('>Q', raw_msg_len)[0]
+    msg_len = struct.unpack(">Q", raw_msg_len)[0]
 
     # Read the message data
     return recv_all(sock, msg_len)
 
 
-def recv_all(sock, n) -> Optional[bytes]:
+def recv_all(sock, n) -> bytes | None:
     # Helper function to recv n bytes or return None if EOF is hit
     data = bytearray()
 

socket__tcp__examples/http__get.py

@@ -1,14 +1,16 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 
-__author__ = 'ipetrash'
+__author__ = "ipetrash"
+
+
+import socket
 
 
 request = b"GET / HTTP/1.1\nHost: stackoverflow.com\n\n"
 host = "stackoverflow.com"
 port = 80
 
-import socket
 sock = socket.socket()
 sock.settimeout(60)  # Если за 60 секунд данные не придут, соединение закрывается
 sock.connect((host, port))

socket__tcp__examples/http__get__via_proxy.py

@@ -1,19 +1,22 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 
-__author__ = 'ipetrash'
+__author__ = "ipetrash"
 
 
 # SOURCE: http://stackoverflow.com/a/42913875/5909792
 # Connect to proxy, in http request write full host address
 
+
+import socket
+
+
 BUFFER_SIZE = 1024
 
 request = b"GET http://stackoverflow.com HTTP/1.1\nHost: stackoverflow.com\n\n"
 host = "proxy.compassplus.ru"
 port = 3128
 
-import socket
 sock = socket.socket()
 sock.settimeout(60)  # Если за 60 секунд данные не придут, соединение закрывается
 

socket__tcp__examples/http__get__with_urlsplit.py

@@ -1,18 +1,20 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 
-__author__ = 'ipetrash'
+__author__ = "ipetrash"
 
 
-from urllib.parse import urlsplit
 import socket
 import re
 
+from urllib.parse import urlsplit
+
+
 socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 # url = input ("Enter a url: ")
-url = 'http://google.ru/'
+url = "http://google.ru/"
 try:
-    re.search("^http[s]?://.*?", url)
+    re.search("^https?://.*?", url)
 except:
     print("Error")
 
@@ -21,14 +23,14 @@
 
 socket.connect((result.netloc, 80))
 
-cmd = f'GET {result.path} HTTP/1.0\r\n\r\n'.encode()
+cmd = f"GET {result.path} HTTP/1.0\r\n\r\n".encode()
 print(cmd)
 socket.send(cmd)
 
 while True:
     data = socket.recv(512)
     if not len(data):
         break
-    print(data, end='')
+    print(data, end="")
 
 socket.close()

socket__tcp__examples/http__post__contact_ng_server.py

@@ -1,11 +1,14 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 
-__author__ = 'ipetrash'
+__author__ = "ipetrash"
 
 
-INT_SOFT_ID = '<INT_SOFT_ID>'
-POINT_CODE = '<POINT_CODE>'
+import socket
+
+
+INT_SOFT_ID = "<INT_SOFT_ID>"
+POINT_CODE = "<POINT_CODE>"
 
 # NG Server
 HOST = "10.7.8.31"
@@ -20,40 +23,41 @@
 SignOut="No"
 ExpectSigned="No"
 />
-""".format(INT_SOFT_ID=INT_SOFT_ID, POINT_CODE=POINT_CODE)
+""".format(
+    INT_SOFT_ID=INT_SOFT_ID, POINT_CODE=POINT_CODE
+)
 
 
 http_request = (
-    'POST / HTTP/1.1\r\n',
-    'Host: {host}:{port}\r\n',
-    'Accept-Encoding: gzip, deflate\r\n',
-    'User-Agent: {user_agent}\r\n',
-    'Connection: keep-alive\r\n',
-    'Accept: */*\r\n',
-    'Content-Length: {content_length}\r\n',
-    '\r\n',
-    '\n',
-    '{body}\n'
+    "POST / HTTP/1.1\r\n",
+    "Host: {host}:{port}\r\n",
+    "Accept-Encoding: gzip, deflate\r\n",
+    "User-Agent: {user_agent}\r\n",
+    "Connection: keep-alive\r\n",
+    "Accept: */*\r\n",
+    "Content-Length: {content_length}\r\n",
+    "\r\n",
+    "\n",
+    "{body}\n",
 )
-http_request = ''.join(http_request)
+http_request = "".join(http_request)
 http_request = http_request.format(
     host=HOST,
     port=PORT,
-    user_agent='iHuman',
+    user_agent="iHuman",
     content_length=len(post_data),
     body=post_data,
 )
 
 print(repr(http_request))
 
 
-import socket
 sock = socket.socket()
 sock.connect((HOST, PORT))
 sock.send(http_request.encode())
 
-print('Socket name: {}'.format(sock.getsockname()))
-print('\nResponse:')
+print("Socket name: {}".format(sock.getsockname()))
+print("\nResponse:")
 
 while True:
     data = sock.recv(1024)

socket__tcp__examples/http_server.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 
-__author__ = 'ipetrash'
+__author__ = "ipetrash"
 
 
 # SOURCE: http://andreymal.org/socket3/
@@ -63,7 +63,7 @@ def parse(conn):  # Обработка соединения в отдельно
 sock.bind(("", PORT))
 sock.listen()
 
-print(f'HTTP server running on http://127.0.0.1:{PORT}/time.html')
+print(f"HTTP server running on http://127.0.0.1:{PORT}/time.html")
 
 try:
     # Работаем постоянно

socket__tcp__examples/http_server__threading.py

@@ -1,13 +1,15 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 
-__author__ = 'ipetrash'
+__author__ = "ipetrash"
 
 
 # SOURCE: http://andreymal.org/socket3/
 
-import time
+
 import socket
+import time
+
 from threading import Thread
 
 
@@ -66,19 +68,19 @@ def parse(conn):  # Обработка соединения в отдельно
         conn.close()
 
 
-HOST = 'localhost'
+HOST = "localhost"
 PORT = 9090
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     sock = socket.socket()
-    print('Socket created')
+    print("Socket created")
 
     sock.bind((HOST, PORT))
-    print('Socket bind complete')
+    print("Socket bind complete")
 
     sock.listen()
-    print('Socket now listening: http://{}:{}'.format(*sock.getsockname()))
+    print("Socket now listening: http://{}:{}".format(*sock.getsockname()))
     print()
 
     try:

socket__tcp__examples/https__get__with_urlsplit.py

@@ -1,32 +1,33 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 
-__author__ = 'ipetrash'
+__author__ = "ipetrash"
 
 
-from urllib.parse import urlsplit
 import socket
 import ssl
 
+from urllib.parse import urlsplit
+
 
 socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-url = 'https://www.coursera.org/robots.txt'
+url = "https://www.coursera.org/robots.txt"
 
 result = urlsplit(url)
 
 context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
 s_sock = context.wrap_socket(socket, server_hostname=result.netloc)
 s_sock.connect((result.netloc, 443))
 
-cmd = f'GET {result.path} HTTP/1.1\r\nHost: {result.netloc}\r\n\r\n'.encode()
+cmd = f"GET {result.path} HTTP/1.1\r\nHost: {result.netloc}\r\n\r\n".encode()
 s_sock.send(cmd)
 
 while True:
     data = s_sock.recv(512)
     if not data:
         break
 
-    print(data.decode(), end='')
+    print(data.decode(), end="")
 
 s_sock.close()