Năm thứ 1: Học Python bằng cách căn chỉnh

Sử dụng nền tảng lập trình hướng đối tượng của Java để vượt qua các khía cạnh quan trọng như kiểu gợi ý (Type Hints), lập trình bất đồng bộ (async/await), và thư viện sinh tạo (Generators & Libraries).

Đêm 1: Học về cú pháp kiểu dữ liệu

# Gán giá trị thông thường và kiểu gợi ý

from typing import *
from dataclasses import dataclass

name: str = "james"
age: int = 25
print("name =", name, ", age =", age)
print(f"name={name}, age={age}")

name: "kobe"
age: int = 26
age = "barent"
print(f"name={name}, age={age}")

names: list[str] = ["james", "kobe", "barten"]
print(names)
print(f"names={names}")

names2: list = ["james", 1, "barten"]
print(names2)
print(f"names={names2}")

ages: dict[str, int] = {"james": 25, "kobe": 26}
print(ages)
print(f"ages={ages}")

scores: set[int] = {85, 90, 95, 90}
print(scores)
print(f"scores={scores}")

names3: None = None
if names3 is not None:
    print(f"names3 is not None, {len(names3)}")
else:
    print("names3 is None")


def name_add(userid: str | int) -> str:
    if isinstance(userid, int):
        return f"userid:{userid}"
    return str(userid)

print(name_add(123))
print(name_add("james"))

def demo():
    length_func: Callable[[str], int] = len
    print(length_func("hello"))

    adder: Callable[[int, int], int] = lambda x, y: x + y
    print(adder(1, 2))


T = TypeVar("T")
K = TypeVar("K")
V = TypeVar("V")

def key_value_pairs(t: tuple[K, V]) -> tuple[K, V]:
    return t
print(key_value_pairs(("a", 1)))

def get_first(t: tuple[T]) -> T | None:
    return t[0] if t else None

print(get_first((1, 2, 3)))
print(get_first(()))

@dataclass
class Pair(Generic[K, V]):
    def __init__(self, key: K, value: V):
        self.key = key
        self.value = value

    def __repr__(self):
        return f"Pair(key={self.key}, value={self.value})"

print(Pair("a", 1))


def demo_protocol():
    class Printable(Protocol):
        def print(self) -> None: ...

    class Dog:
        def __init__(self, name: str):
            self.name = name
        def print(self) -> None:
            print(f"Dog: {self.name}")

    class Cat:
        def __init__(self, name: str):
            self.name = name
        def print(self) -> None:
            print(f"Cat: {self.name}")

    def print_animal(animal: Printable) -> None:
        animal.print()

    print_animal(Dog("Buddy"))
    print_animal(Cat("Whiskers"))

demo_protocol()



def demo_dataclass():
    @dataclass
    class Point:
        x: int
        y: int

    p = Point(1, 2)
    print(p)
    print(f"Point: ({p.x}, {p.y})")
demo_dataclass()

def demo_type_alias():
    UserId = str
    ScoreMap = dict[str, int]
    Handler = Callable[[dict[str, Any]], None] 

    def get_user(uid: UserId) -> UserId:
        return f"UserId-{uid}"

    scores: ScoreMap = {"james": 85, "kobe": 90}
    handler: Handler = lambda scores: print(f"scores={scores}")

    print(get_user("james"))
    handler(scores)
demo_type_alias()

Khi so sánh với Java, ngôn ngữ lập trình này không bắt buộc phải chỉ định kiểu dữ liệu cho mỗi biến, điều này tạo ra sự khác biệt đáng chú ý.

Đêm 2: Luồng và lập trình bất đồng bộ

import asyncio
import random
import time
import threading
from typing import *

# Cơ bản về coroutines
async def greet(name: str, wait_seconds: int = 10) -> str:
    """Hàm chào gọi trong chế độ bất đồng bộ, mô phỏng tác vụ thời gian thực với đếm ngược"""
    for remaining in range(wait_seconds, 0, -1):
        print(f"  ⏳ Đang chờ... Còn lại {remaining} giây")
        await asyncio.sleep(1)  
    print(f"  Thread hiện tại: {threading.current_thread().name}")
    return f"Chào bạn, {name}!"


async def greet2(name: str, wait_seconds: int = 5) -> str:
    """Một hàm chào gọi khác cũng có đếm ngược"""
    for remaining in range(wait_seconds, 0, -1):
        print(f"  ⏳ greet2[{name}] Đang chờ... Còn lại {remaining} giây")
        await asyncio.sleep(1)  
    print(f"  greet2[{name}] Hoàn thành! Thread: {threading.current_thread().name}")
    return f"Xin chào, {name}!"

async def hello():
    """Hàm chính: Gọi đồng bộ greet2 và greet, thể hiện việc chạy song song"""
    print("Chào thế giới")
    task2 = asyncio.create_task(greet2("Java", wait_seconds=5))
    result1 = await greet("Python", wait_seconds=10)
    print(f"  Kết quả greet: {result1}")
    result2 = await task2
    print(f"  Kết quả greet2: {result2}")
    print(f"Thread hiện tại: {threading.current_thread().name}")

asyncio.run(hello())

# Thực hiện song song
async def demo_gather():
    """Demos asyncio.gather để thực hiện nhiều coroutines cùng một lúc"""
    async def fetch(url: str) -> str:
        print(f"Fetching {url}")
        if url == "http://www.google.com":
            await asyncio.sleep(1)  
            raise ValueError("Lỗi kiểm tra")
        else:
            await asyncio.sleep(3)  
            print(f"Fetching {url} hoàn thành")
            return f"Kết quả từ {url}"
    try:
        await asyncio.gather(
            fetch("http://www.google.com"),
            fetch("http://www.bing.com"),
            fetch("http://www.yahoo.com")
        )
    except ValueError as e:
        print(f"Catch exception: {e}")
    
    print("Gather đã trả về, tiếp tục làm công việc khác...")
    await asyncio.sleep(5)  
    print("Hàm chính kết thúc")

asyncio.run(demo_gather())

# Lập trình bất đồng bộ cấu trúc hóa
async def demo_task_group():
    """Demos asyncio.TaskGroup để lập trình bất đồng bộ cấu trúc hóa"""
    async def fetch(url: str) -> str:
        print(f"Fetching {url}")
        if url == "http://www.google.com":
            await asyncio.sleep(1)  
            raise ValueError("Lỗi kiểm tra")
        else:
            await asyncio.sleep(3)  
            print(f"Fetching {url} hoàn thành")
            return f"Kết quả từ {url}"
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(fetch("http://www.google.com"))
            group.create_task(fetch("http://www.bing.com"))
            group.create_task(fetch("http://www.yahoo.com"))
    except ExceptionGroup as e:
        print(f"Catch exception group: {e}")
    
    print("TaskGroup đã an toàn thoát, không còn nhiệm vụ nào")
    await asyncio.sleep(5)
    print("Hàm chính kết thúc")

    start = time.perf_counter()
    async with asyncio.TaskGroup() as group:
        t1 = group.create_task(fetch("http://www.google.com"))
        t2 = group.create_task(fetch("http://www.bing.com"))
        t3 = group.create_task(fetch("http://www.yahoo.com"))
    end = time.perf_counter()
    print(f"Kết quả: {t1.result()}, {t2.result()}, {t3.result()}")
    print(f"Thời gian trôi qua: {end - start} giây")
    
asyncio.run(demo_task_group())
# An toàn

# Lặp đi lặp lại bất đồng bộ (stream())
async def demo_async_iterator():
    async def async_range(n: int):
        for i in range(n):
            await asyncio.sleep(1)
            yield i  

    print("Kết quả lặp bất đồng bộ:", end="", flush=True)
    async for i in async_range(5):
        print(i, end=" ", flush=True)  
    print()

asyncio.run(demo_async_iterator())


# Quản lý ngữ cảnh bất đồng bộ
async def demo_async_context_manager():
    """ __aenter__ / __aexit__ là phiên bản bất đồng bộ của __enter__ / __exit__
        So sánh với phương thức close() của Java AutoCloseable.close() """
    class AsyncDBConnection:
        async def __aenter__(self):
            print("Mở kết nối cơ sở dữ liệu")
            return self
        async def __aexit__(self, exc_type, exc_value, traceback):
            print("Đóng kết nối cơ sở dữ liệu")
        async def query(self, sql):
            print(f"Thực thi câu lệnh SQL: {sql}")
            await asyncio.sleep(1)
            return [1, 2, 3]
    async with AsyncDBConnection() as db:
        result = await db.query("SELECT * FROM users")
        print(result)

asyncio.run(demo_async_context_manager())

# Kiểm soát thời gian chờ đợi -- So sánh với Future.get(timeout) của Java
async def demo_timeout_control():
    pass